fix(csharp): address PR #375 review feedback

- Fix maxRetries semantics: -1=unlimited, 0=no retries, >0=max retries.
  Move check to catch block so first attempt always runs.
- Use real elapsed time (stopwatch) for retry budget instead of summing
  backoff delays, so slow/hanging attempts count against the budget.
- Use long for retryTimeoutMs to prevent int overflow on large values.
- Remove raw exception messages from trace events to avoid leaking
  pre-signed URL credentials in telemetry.
- Throw on invalid max_retries property values instead of silent fallback.
- Replace flaky Task.Delay(3000) in test with deterministic polling.

Co-authored-by: Isaac

Author: eric-wang-1990

csharp/src/Reader/CloudFetch/CloudFetchConfiguration.cs

@@ -125,6 +125,20 @@ public static CloudFetchConfiguration FromProperties(
             Schema schema,
             bool isLz4Compressed)
         {
+            // Parse MaxRetries separately: -1 (default) = not set, >= 0 = explicit limit.
+            // Throw on non-integer values to surface misconfiguration.
+            int parsedMaxRetries = DefaultMaxRetries;
+            if (properties.TryGetValue(DatabricksParameters.CloudFetchMaxRetries, out string? maxRetriesStr))
+            {
+                if (!int.TryParse(maxRetriesStr, out int maxRetries) || maxRetries < -1)
+                {
+                    throw new ArgumentException(
+                        $"Invalid value '{maxRetriesStr}' for {DatabricksParameters.CloudFetchMaxRetries}. " +
+                        $"Expected -1 (no limit) or a non-negative integer.");
+                }
+                parsedMaxRetries = maxRetries;
+            }
+
             var config = new CloudFetchConfiguration
             {
                 Schema = schema ?? throw new ArgumentNullException(nameof(schema)),
@@ -133,7 +147,7 @@ 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 = properties.TryGetValue(DatabricksParameters.CloudFetchMaxRetries, out string? maxRetriesStr) && int.TryParse(maxRetriesStr, out int maxRetries) && maxRetries >= 0 ? maxRetries : DefaultMaxRetries,
+                MaxRetries = parsedMaxRetries,
                 RetryTimeoutSeconds = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchRetryTimeoutSeconds, DefaultRetryTimeoutSeconds),
                 RetryDelayMs = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchRetryDelayMs, DefaultRetryDelayMs),
                 MaxUrlRefreshAttempts = PropertyHelper.GetPositiveIntPropertyWithValidation(properties, DatabricksParameters.CloudFetchMaxUrlRefreshAttempts, DefaultMaxUrlRefreshAttempts),

csharp/src/Reader/CloudFetch/CloudFetchDownloader.cs

@@ -116,7 +116,7 @@ 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 (0 = no limit, use timeout only).</param>
+        /// <param name="maxRetries">Maximum retry attempts. -1 = no limit (use timeout only), 0 = no retries (single attempt), positive = max retry 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(
@@ -492,30 +492,16 @@ await _activityTracer.TraceActivityAsync(async activity =>
             ]);
 
                 // Retry logic with time-budget approach and exponential backoff with jitter.
-                // Similar to RetryHttpHandler: keeps retrying until the time budget is exhausted,
+                // Keeps retrying until the time budget (real elapsed time) is exhausted,
                 // rather than a fixed retry count. This gives transient issues (firewall, proxy 502,
                 // connection drops) enough time to resolve.
                 int currentBackoffMs = _retryDelayMs;
-                int totalRetryWaitMs = 0;
+                long retryTimeoutMs = Math.Min((long)_retryTimeoutSeconds, int.MaxValue / 1000L) * 1000L;
                 int attemptCount = 0;
                 Exception? lastException = null;
 
                 while (!cancellationToken.IsCancellationRequested)
                 {
-                    // Check if we've exceeded the max retry count (if set)
-                    if (_maxRetries >= 0 && attemptCount >= _maxRetries)
-                    {
-                        activity?.AddEvent("cloudfetch.download_max_retries_exceeded", [
-                            new("offset", downloadResult.StartRowOffset),
-                            new("sanitized_url", SanitizeUrl(url)),
-                            new("total_attempts", attemptCount),
-                            new("max_retries", _maxRetries),
-                            new("last_error", lastException?.GetType().Name ?? "none"),
-                            new("last_error_message", lastException?.Message ?? "none")
-                        ]);
-                        break;
-                    }
-
                     attemptCount++;
                     try
                     {
@@ -594,36 +580,43 @@ await _activityTracer.TraceActivityAsync(async activity =>
                     {
                         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've exceeded the max retry count (if set)
+                        // -1 = unlimited, 0 = no retries (single attempt), >0 = max retry attempts
+                        if (_maxRetries >= 0 && attemptCount > _maxRetries)
+                        {
+                            activity?.AddEvent("cloudfetch.download_max_retries_exceeded", [
+                                new("offset", downloadResult.StartRowOffset),
+                                new("sanitized_url", SanitizeUrl(url)),
+                                new("total_attempts", attemptCount),
+                                new("max_retries", _maxRetries)
+                            ]);
+                            break;
+                        }
 
-                        // Check if we would exceed the retry time budget
-                        int retryTimeoutMs = _retryTimeoutSeconds * 1000;
-                        if (retryTimeoutMs > 0 && totalRetryWaitMs + waitMs > retryTimeoutMs)
+                        // Check if we've exceeded the time budget (real elapsed time)
+                        if (retryTimeoutMs > 0 && stopwatch.ElapsedMilliseconds >= retryTimeoutMs)
                         {
                             activity?.AddEvent("cloudfetch.download_retry_timeout_exceeded", [
                                 new("offset", downloadResult.StartRowOffset),
                                 new("sanitized_url", SanitizeUrl(url)),
                                 new("total_attempts", attemptCount),
-                                new("total_retry_wait_ms", totalRetryWaitMs),
+                                new("elapsed_ms", stopwatch.ElapsedMilliseconds),
                                 new("retry_timeout_seconds", _retryTimeoutSeconds),
-                                new("last_error", ex.GetType().Name),
-                                new("last_error_message", ex.Message)
+                                new("last_error", ex.GetType().Name)
                             ]);
-                            break; // Exceeded time budget
+                            break;
                         }
 
-                        totalRetryWaitMs += waitMs;
+                        // Exponential backoff with jitter (80-120% of base)
+                        int waitMs = (int)Math.Max(100, currentBackoffMs * (0.8 + new Random().NextDouble() * 0.4));
 
                         activity?.AddEvent("cloudfetch.download_retry", [
-                            new("error.context", "cloudfetch.download_retry"),
                             new("offset", downloadResult.StartRowOffset),
                             new("sanitized_url", SanitizeUrl(url)),
                             new("attempt", attemptCount),
+                            new("elapsed_ms", stopwatch.ElapsedMilliseconds),
                             new("retry_timeout_seconds", _retryTimeoutSeconds),
-                            new("total_retry_wait_ms", totalRetryWaitMs),
                             new("error_type", ex.GetType().Name),
-                            new("error_message", ex.Message),
                             new("backoff_ms", waitMs)
                         ]);
 
@@ -639,7 +632,6 @@ await _activityTracer.TraceActivityAsync(async activity =>
                         new("offset", downloadResult.StartRowOffset),
                         new("sanitized_url", sanitizedUrl),
                         new("total_attempts", attemptCount),
-                        new("total_retry_wait_ms", totalRetryWaitMs),
                         new("elapsed_time_ms", stopwatch.ElapsedMilliseconds)
                     ]);
 
@@ -649,7 +641,7 @@ await _activityTracer.TraceActivityAsync(async activity =>
                         ? $"max_retries: {_maxRetries}, timeout: {_retryTimeoutSeconds}s"
                         : $"timeout: {_retryTimeoutSeconds}s";
                     throw new InvalidOperationException(
-                        $"Failed to download file from {sanitizedUrl} after {attemptCount} attempts over {totalRetryWaitMs / 1000}s ({retryLimits}).",
+                        $"Failed to download file from {sanitizedUrl} after {attemptCount} attempts over {stopwatch.ElapsedMilliseconds / 1000}s ({retryLimits}).",
                         lastException);
                 }
 

csharp/test/E2E/CloudFetch/CloudFetchDownloaderTest.cs

@@ -271,16 +271,22 @@ public async Task DownloadFileAsync_HandlesHttpError_AndSetsFailedOnDownloadResu
             await downloader.StartAsync(CancellationToken.None);
             _downloadQueue.Add(mockDownloadResult.Object);
 
-            // Wait for retries to exhaust the time budget
-            await Task.Delay(3000);
-
             // 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<InvalidOperationException>(capturedException);
 
             // Verify the downloader has an error