feat(csharp): convert cloudfetch retry from limit by # of retries to limit by time (#375)

## Summary
- Replace fixed 3-attempt retry with **5-minute time-budget** approach
using exponential backoff with jitter (matching RetryHttpHandler's
pattern)
- Retry all exceptions (502 Bad Gateway from proxy, connection drops,
TCP resets, HttpClient.Timeout) — catch-all, not just transport errors
- Add new `adbc.databricks.cloudfetch.retry_timeout_seconds` parameter
(default 300s / 5 min)
- Keep `adbc.databricks.cloudfetch.max_retries` parameter for PowerBI
backward compatibility (default -1 = not set). When set, retry loop
exits if either max retries or timeout is reached first
- Add detailed retry tracing (attempt count, backoff delay, timeout
budget remaining)

### Problem
CloudFetch downloads used a fixed 3-retry loop with linear backoff
(~1.5s total). When a proxy returns 502 or a firewall blocks cloud
storage, all 3 retries exhaust within seconds — not enough time for
transient issues to resolve.

### Solution
Switch to a time-budget approach (same as RetryHttpHandler from #373):
keep retrying with exponential backoff + jitter until the 5-minute
budget is exhausted. This gives transient issues (proxy errors, firewall
rules, connection drops) enough time to resolve.

| | Before | After |
|---|---|---|
| Retry limit | 3 attempts | 5-minute time budget (+ optional
max_retries) |
| Backoff | Linear (500ms, 1s, 1.5s) | Exponential with jitter (500ms →
1s → 2s → ... → 32s cap) |
| Total retry window | ~3 seconds | Up to 5 minutes |
| Error classification | Retry all exceptions | Retry all exceptions
(same) |
| max_retries param | Required (default 3) | Optional (default -1 = not
set) |

## Test plan
- [x] All 9 CloudFetch downloader unit tests pass
- [x] Full build succeeds (netstandard2.0 + net8.0)
- [ ] Manual test: block cloud storage via MITM firewall, verify retries
continue for 5 min
- [ ] Manual test: unblock within 5 min, verify download resumes
successfully

This pull request was AI-assisted by Isaac.

Author: eric-wang-1990

csharp/src/DatabricksParameters.cs

@@ -52,11 +52,19 @@ public class DatabricksParameters : SparkParameters
         public const string MaxBytesPerFile = "adbc.databricks.cloudfetch.max_bytes_per_file";
 
         /// <summary>
-        /// Maximum number of retry attempts for CloudFetch downloads.
-        /// Default value is 3 if not specified.
+        /// Maximum number of retry attempts for CloudFetch downloads (total attempts, including first try).
+        /// Default value is 0 (no limit, use timeout only).
+        /// When set to a positive value, the retry loop exits if either this count or the timeout is reached.
         /// </summary>
         public const string CloudFetchMaxRetries = "adbc.databricks.cloudfetch.max_retries";
 
+        /// <summary>
+        /// Maximum time in seconds to retry failed CloudFetch downloads.
+        /// Uses exponential backoff with jitter within this time budget.
+        /// Default value is 300 (5 minutes) if not specified.
+        /// </summary>
+        public const string CloudFetchRetryTimeoutSeconds = "adbc.databricks.cloudfetch.retry_timeout_seconds";
+
         /// <summary>
         /// Delay in milliseconds between CloudFetch retry attempts.
         /// Default value is 500ms if not specified.

csharp/src/Reader/CloudFetch/CloudFetchConfiguration.cs

@@ -33,7 +33,8 @@ internal sealed class CloudFetchConfiguration
         internal const int DefaultPrefetchCount = 2;
         internal const int DefaultMemoryBufferSizeMB = 200;
         internal const int DefaultTimeoutMinutes = 5;
-        internal const int DefaultMaxRetries = 3;
+        internal const int DefaultMaxRetries = 0; // 0 = no limit (use timeout only)
+        internal const int DefaultRetryTimeoutSeconds = 300; // 5 minutes
         internal const int DefaultRetryDelayMs = 500;
         internal const int DefaultMaxUrlRefreshAttempts = 3;
         internal const int DefaultUrlExpirationBufferSeconds = 60;
@@ -59,10 +60,18 @@ internal sealed class CloudFetchConfiguration
         public int TimeoutMinutes { get; set; } = DefaultTimeoutMinutes;
 
         /// <summary>
-        /// Maximum retry attempts for failed downloads.
+        /// Maximum retry attempts for failed downloads (total attempts, including first try).
+        /// 0 means no limit (use timeout only). When set to a positive value,
+        /// the retry loop exits if either this count or the timeout is reached.
         /// </summary>
         public int MaxRetries { get; set; } = DefaultMaxRetries;
 
+        /// <summary>
+        /// Maximum time in seconds to retry failed downloads before giving up.
+        /// Uses exponential backoff with jitter within this time budget.
+        /// </summary>
+        public int RetryTimeoutSeconds { get; set; } = DefaultRetryTimeoutSeconds;
+
         /// <summary>
         /// Delay between retry attempts (in milliseconds).
         /// </summary>
@@ -116,6 +125,20 @@ public static CloudFetchConfiguration FromProperties(
             Schema schema,
             bool isLz4Compressed)
         {
+            // Parse MaxRetries separately: 0 (default) = no limit, >0 = total attempt count.
+            // Throw on non-integer or negative values to surface misconfiguration.
+            int parsedMaxRetries = DefaultMaxRetries;
+            if (properties.TryGetValue(DatabricksParameters.CloudFetchMaxRetries, out string? maxRetriesStr))
+            {
+                if (!int.TryParse(maxRetriesStr, out int maxRetries) || maxRetries < 0)
+                {
+                    throw new ArgumentException(
+                        $"Invalid value '{maxRetriesStr}' for {DatabricksParameters.CloudFetchMaxRetries}. " +
+                        $"Expected 0 (no limit) or a positive integer.");
+                }
+                parsedMaxRetries = maxRetries;
+            }
+
             var config = new CloudFetchConfiguration
             {
                 Schema = schema ?? throw new ArgumentNullException(nameof(schema)),
@@ -124,7 +147,8 @@ public static CloudFetchConfiguration FromProperties(
                 PrefetchCount = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchPrefetchCount, DefaultPrefetchCount),
                 MemoryBufferSizeMB = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchMemoryBufferSize, DefaultMemoryBufferSizeMB),
                 TimeoutMinutes = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchTimeoutMinutes, DefaultTimeoutMinutes),
-                MaxRetries = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchMaxRetries, DefaultMaxRetries),
+                MaxRetries = parsedMaxRetries,
+                RetryTimeoutSeconds = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchRetryTimeoutSeconds, DefaultRetryTimeoutSeconds),
                 RetryDelayMs = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchRetryDelayMs, DefaultRetryDelayMs),
                 MaxUrlRefreshAttempts = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchMaxUrlRefreshAttempts, DefaultMaxUrlRefreshAttempts),
                 UrlExpirationBufferSeconds = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchUrlExpirationBufferSeconds, DefaultUrlExpirationBufferSeconds)

csharp/src/Reader/CloudFetch/CloudFetchDownloader.cs

@@ -51,6 +51,7 @@ internal sealed class CloudFetchDownloader : ICloudFetchDownloader
         private readonly int _maxParallelDownloads;
         private readonly bool _isLz4Compressed;
         private readonly int _maxRetries;
+        private readonly int _retryTimeoutSeconds;
         private readonly int _retryDelayMs;
         private readonly int _maxUrlRefreshAttempts;
         private readonly int _urlExpirationBufferSeconds;
@@ -94,6 +95,7 @@ public CloudFetchDownloader(
             _maxParallelDownloads = config.ParallelDownloads;
             _isLz4Compressed = config.IsLz4Compressed;
             _maxRetries = config.MaxRetries;
+            _retryTimeoutSeconds = config.RetryTimeoutSeconds;
             _retryDelayMs = config.RetryDelayMs;
             _maxUrlRefreshAttempts = config.MaxUrlRefreshAttempts;
             _urlExpirationBufferSeconds = config.UrlExpirationBufferSeconds;
@@ -114,8 +116,9 @@ public CloudFetchDownloader(
         /// <param name="resultFetcher">The result fetcher that manages URLs.</param>
         /// <param name="maxParallelDownloads">Maximum parallel downloads.</param>
         /// <param name="isLz4Compressed">Whether results are LZ4 compressed.</param>
-        /// <param name="maxRetries">Maximum retry attempts (optional, default 3).</param>
-        /// <param name="retryDelayMs">Delay between retries in ms (optional, default 1000).</param>
+        /// <param name="maxRetries">Total number of attempts. 0 = no limit (use timeout only), positive = max total attempts.</param>
+        /// <param name="retryTimeoutSeconds">Time budget for retries in seconds (optional, default 300).</param>
+        /// <param name="retryDelayMs">Initial delay between retries in ms (optional, default 500).</param>
         internal CloudFetchDownloader(
             IActivityTracer activityTracer,
             BlockingCollection<IDownloadResult> downloadQueue,
@@ -125,8 +128,9 @@ internal CloudFetchDownloader(
             ICloudFetchResultFetcher resultFetcher,
             int maxParallelDownloads,
             bool isLz4Compressed,
-            int maxRetries = 3,
-            int retryDelayMs = 1000)
+            int maxRetries = CloudFetchConfiguration.DefaultMaxRetries,
+            int retryTimeoutSeconds = CloudFetchConfiguration.DefaultRetryTimeoutSeconds,
+            int retryDelayMs = CloudFetchConfiguration.DefaultRetryDelayMs)
         {
             _activityTracer = activityTracer ?? throw new ArgumentNullException(nameof(activityTracer));
             _downloadQueue = downloadQueue ?? throw new ArgumentNullException(nameof(downloadQueue));
@@ -138,6 +142,7 @@ internal CloudFetchDownloader(
             _maxParallelDownloads = maxParallelDownloads;
             _isLz4Compressed = isLz4Compressed;
             _maxRetries = maxRetries;
+            _retryTimeoutSeconds = retryTimeoutSeconds;
             _retryDelayMs = retryDelayMs;
             _maxUrlRefreshAttempts = CloudFetchConfiguration.DefaultMaxUrlRefreshAttempts;
             _urlExpirationBufferSeconds = CloudFetchConfiguration.DefaultUrlExpirationBufferSeconds;
@@ -486,9 +491,31 @@ await _activityTracer.TraceActivityAsync(async activity =>
                     new("expected_size_kb", size / 1024.0)
             ]);
 
-                // Retry logic for downloading files
-                for (int retry = 0; retry < _maxRetries; retry++)
+                // Retry logic with time-budget approach and exponential backoff with jitter.
+                // Same pattern as RetryHttpHandler: tracks cumulative backoff sleep time against
+                // the budget. This gives transient issues (firewall, proxy 502, connection drops)
+                // enough time to resolve.
+                int currentBackoffMs = (int)Math.Min(Math.Max(0L, (long)_retryDelayMs), 32_000L);
+                long retryTimeoutMs = Math.Min((long)_retryTimeoutSeconds, int.MaxValue / 1000L) * 1000L;
+                long totalRetryWaitMs = 0;
+                int attemptCount = 0;
+                Exception? lastException = null;
+
+                while (!cancellationToken.IsCancellationRequested)
                 {
+                    // Check max retry count before each attempt (0 = no limit, >0 = total attempts)
+                    if (_maxRetries > 0 && attemptCount >= _maxRetries)
+                    {
+                        activity?.AddEvent("cloudfetch.download_max_retries_exceeded", [
+                            new("offset", downloadResult.StartRowOffset),
+                            new("sanitized_url", sanitizedUrl),
+                            new("total_attempts", attemptCount),
+                            new("max_retries", _maxRetries)
+                        ]);
+                        break;
+                    }
+
+                    attemptCount++;
                     try
                     {
                         // Create HTTP request with optional custom headers
@@ -562,18 +589,41 @@ await _activityTracer.TraceActivityAsync(async activity =>
                         fileData = await response.Content.ReadAsByteArrayAsync().ConfigureAwait(false);
                         break; // Success, exit retry loop
                     }
-                    catch (Exception ex) when (retry < _maxRetries - 1 && !cancellationToken.IsCancellationRequested)
+                    catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
                     {
-                        // Log the error and retry
-                        activity?.AddException(ex, [
-                            new("error.context", "cloudfetch.download_retry"),
+                        lastException = ex;
+
+                        // Exponential backoff with jitter (80-120% of base)
+                        int waitMs = (int)Math.Max(100, currentBackoffMs * (0.8 + new Random().NextDouble() * 0.4));
+
+                        // Check if we would exceed the time budget
+                        if (retryTimeoutMs > 0 && totalRetryWaitMs + waitMs > retryTimeoutMs)
+                        {
+                            activity?.AddEvent("cloudfetch.download_retry_timeout_exceeded", [
+                                new("offset", downloadResult.StartRowOffset),
+                                new("sanitized_url", sanitizedUrl),
+                                new("total_attempts", attemptCount),
+                                new("total_retry_wait_ms", totalRetryWaitMs),
+                                new("retry_timeout_seconds", _retryTimeoutSeconds),
+                                new("last_error", ex.GetType().Name)
+                            ]);
+                            break;
+                        }
+
+                        totalRetryWaitMs += waitMs;
+
+                        activity?.AddEvent("cloudfetch.download_retry", [
                             new("offset", downloadResult.StartRowOffset),
-                            new("sanitized_url", SanitizeUrl(url)),
-                            new("attempt", retry + 1),
-                            new("max_retries", _maxRetries)
+                            new("sanitized_url", sanitizedUrl),
+                            new("attempt", attemptCount),
+                            new("total_retry_wait_ms", totalRetryWaitMs),
+                            new("retry_timeout_seconds", _retryTimeoutSeconds),
+                            new("error_type", ex.GetType().Name),
+                            new("backoff_ms", waitMs)
                         ]);
 
-                        await Task.Delay(_retryDelayMs * (retry + 1), cancellationToken).ConfigureAwait(false);
+                        await Task.Delay(waitMs, cancellationToken).ConfigureAwait(false);
+                        currentBackoffMs = (int)Math.Min((long)currentBackoffMs * 2L, 32_000L);
                     }
                 }
 
@@ -583,13 +633,18 @@ await _activityTracer.TraceActivityAsync(async activity =>
                     activity?.AddEvent("cloudfetch.download_failed_all_retries", [
                         new("offset", downloadResult.StartRowOffset),
                         new("sanitized_url", sanitizedUrl),
-                        new("max_retries", _maxRetries),
+                        new("total_attempts", attemptCount),
+                        new("total_retry_wait_ms", totalRetryWaitMs),
                         new("elapsed_time_ms", stopwatch.ElapsedMilliseconds)
                     ]);
 
                     // Release the memory we acquired
                     _memoryManager.ReleaseMemory(size);
-                    throw new InvalidOperationException($"Failed to download file from {url} after {_maxRetries} attempts.");
+                    string retryLimits = _maxRetries > 0
+                        ? $"max_retries: {_maxRetries}, timeout: {_retryTimeoutSeconds}s"
+                        : $"timeout: {_retryTimeoutSeconds}s";
+                    throw new InvalidOperationException(
+                        $"Failed to download file from {sanitizedUrl} after {attemptCount} attempts over {stopwatch.Elapsed.TotalSeconds:F1}s ({retryLimits}). Last error: {lastException?.GetType().Name ?? "unknown"}");
                 }
 
                 // Process the downloaded file data

csharp/test/E2E/CloudFetch/CloudFetchDownloaderTest.cs

@@ -178,8 +178,9 @@ public async Task DownloadFileAsync_ProcessesFile_AndAddsToResultQueue()
                 _mockResultFetcher.Object,
                 1, // maxParallelDownloads
                 false, // isLz4Compressed
-                1, // maxRetries
-                10); // retryDelayMs
+                maxRetries: 0,
+                retryTimeoutSeconds: 5,
+                retryDelayMs: 10);
 
             // Act
             await downloader.StartAsync(CancellationToken.None);
@@ -262,24 +263,31 @@ public async Task DownloadFileAsync_HandlesHttpError_AndSetsFailedOnDownloadResu
                 _mockResultFetcher.Object,
                 1, // maxParallelDownloads
                 false, // isLz4Compressed
-                1, // maxRetries
-                10); // retryDelayMs
+                maxRetries: 0,
+                retryTimeoutSeconds: 1,
+                retryDelayMs: 10);
 
             // Act
             await downloader.StartAsync(CancellationToken.None);
             _downloadQueue.Add(mockDownloadResult.Object);
 
-            // Wait for the download to be processed
-            await Task.Delay(100);
-
             // Add the end of results guard to complete the downloader
             _downloadQueue.Add(EndOfResultsGuard.Instance);
 
+            // Wait for the download to fail - poll deterministically instead of fixed delay
+            int maxWaitMs = 5000;
+            int waitedMs = 0;
+            while (capturedException == null && waitedMs < maxWaitMs)
+            {
+                await Task.Delay(50);
+                waitedMs += 50;
+            }
+
             // Assert
             // Verify SetFailed was called
+            Assert.True(capturedException != null, $"SetFailed was not called within {maxWaitMs}ms");
             mockDownloadResult.Verify(r => r.SetFailed(It.IsAny<Exception>()), Times.Once);
-            Assert.NotNull(capturedException);
-            Assert.IsType<HttpRequestException>(capturedException);
+            Assert.IsType<InvalidOperationException>(capturedException);
 
             // Verify the downloader has an error
             Assert.True(downloader.HasError);
@@ -351,8 +359,9 @@ public async Task DownloadFileAsync_WithError_StopsProcessingRemainingFiles()
                 _mockResultFetcher.Object,
                 1, // maxParallelDownloads
                 false, // isLz4Compressed
-                1, // maxRetries
-                10); // retryDelayMs
+                maxRetries: 0,
+                retryTimeoutSeconds: 1,
+                retryDelayMs: 10);
 
             // Act
             await downloader.StartAsync(CancellationToken.None);
@@ -365,7 +374,7 @@ public async Task DownloadFileAsync_WithError_StopsProcessingRemainingFiles()
             Console.WriteLine("Added end guard");
 
             // Wait for the download to fail - use a timeout to avoid hanging
-            int maxWaitMs = 2000;
+            int maxWaitMs = 5000;
             int waitedMs = 0;
             while (!downloader.HasError && !setFailedCalled && waitedMs < maxWaitMs)
             {
@@ -684,8 +693,9 @@ public async Task DownloadFileAsync_RefreshesExpiredUrl_WhenHttpErrorOccurs()
                 _mockResultFetcher.Object,
                 1, // maxParallelDownloads
                 false, // isLz4Compressed
-                2, // maxRetries
-                10); // retryDelayMs
+                maxRetries: 0,
+                retryTimeoutSeconds: 5,
+                retryDelayMs: 10);
 
             // Act
             await downloader.StartAsync(CancellationToken.None);