[PECOBLR-3984] Add jitter to chunk download retries

NEXT_CHANGELOG.md

@@ -8,6 +8,8 @@
 - `DatabaseMetaData.getColumns(...)` with a `null` catalog now issues a single `SHOW COLUMNS IN ALL CATALOGS` statement (consistent with `getSchemas`/`getTables`) instead of enumerating every catalog and issuing a per-catalog `SHOW COLUMNS`. Older DBR versions that do not support the syntax transparently fall back to the previous enumerate-and-fan-out behavior.
 
 ### Fixed
+- Added jitter to chunk-download retries to reduce synchronized retry bursts during transient failures without increasing the existing maximum delay.
+
 - Fixed `IdleConnectionEvictor` thread leak in long-running applications. Driver-side resources (HTTP client, background threads) are now always released when `Connection.close()` is called, even if statement cleanup or server-side session termination fails.
 
 - Throw `DatabricksSQLException` instead of an unchecked `ClassCastException` when a complex-type getter (`getArray`, `getStruct`, `getMap`) is called on a column of a different complex type.

src/main/java/com/databricks/jdbc/api/impl/arrow/ChunkDownloadTask.java

@@ -17,26 +17,36 @@ class ChunkDownloadTask implements DatabricksCallableTask {
 
   private static final JdbcLogger LOGGER = JdbcLoggerFactory.getLogger(ChunkDownloadTask.class);
   public static final int MAX_RETRIES = 5;
-  private static final long RETRY_DELAY_MS = 1500; // 1.5 seconds
   private final ArrowResultChunk chunk;
   private final IDatabricksHttpClient httpClient;
   private final ChunkDownloadManager chunkDownloader;
   private final IDatabricksConnectionContext connectionContext;
   private final String statementId;
   private final ChunkLinkDownloadService<ArrowResultChunk> linkDownloadService;
+  private final ChunkRetryPolicy retryPolicy;
   Throwable uncaughtException = null;
 
   ChunkDownloadTask(
       ArrowResultChunk chunk,
       IDatabricksHttpClient httpClient,
       ChunkDownloadManager chunkDownloader,
       ChunkLinkDownloadService<ArrowResultChunk> linkDownloadService) {
+    this(chunk, httpClient, chunkDownloader, linkDownloadService, new ChunkRetryPolicy());
+  }
+
+  ChunkDownloadTask(
+      ArrowResultChunk chunk,
+      IDatabricksHttpClient httpClient,
+      ChunkDownloadManager chunkDownloader,
+      ChunkLinkDownloadService<ArrowResultChunk> linkDownloadService,
+      ChunkRetryPolicy retryPolicy) {
     this.chunk = chunk;
     this.httpClient = httpClient;
     this.chunkDownloader = chunkDownloader;
     this.connectionContext = DatabricksThreadContextHolder.getConnectionContext();
     this.statementId = DatabricksThreadContextHolder.getStatementId();
     this.linkDownloadService = linkDownloadService;
+    this.retryPolicy = retryPolicy;
   }
 
   @Override
@@ -97,13 +107,14 @@ public Void call() throws DatabricksSQLException, ExecutionException, Interrupte
                 chunk.getChunkIndex(),
                 DatabricksDriverErrorCode.CHUNK_DOWNLOAD_ERROR.name());
           } else {
+            long retryDelayMs = retryPolicy.getRetryDelayMs();
             LOGGER.warn(
                 String.format(
-                    "Retry attempt %d for chunk index: %d, Error: %s",
-                    retries, chunk.getChunkIndex(), e.getMessage()));
+                    "Retry attempt %d for chunk index: %d, retryDelayMs: %d, Error: %s",
+                    retries, chunk.getChunkIndex(), retryDelayMs, e.getMessage()));
             chunk.setStatus(ChunkStatus.DOWNLOAD_RETRY);
             try {
-              Thread.sleep(RETRY_DELAY_MS);
+              retryPolicy.sleep(retryDelayMs);
             } catch (InterruptedException ie) {
               Thread.currentThread().interrupt();
               throw new DatabricksSQLException(

src/main/java/com/databricks/jdbc/api/impl/arrow/ChunkRetryPolicy.java

@@ -0,0 +1,29 @@
+package com.databricks.jdbc.api.impl.arrow;
+
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.LongUnaryOperator;
+
+/** Adds equal jitter so concurrent chunk downloads do not retry in lockstep. */
+final class ChunkRetryPolicy {
+
+  static final long MIN_RETRY_DELAY_MS = 750;
+  static final long MAX_RETRY_DELAY_MS = 1500;
+  private final LongUnaryOperator randomLong;
+
+  ChunkRetryPolicy() {
+    this(bound -> ThreadLocalRandom.current().nextLong(bound));
+  }
+
+  ChunkRetryPolicy(LongUnaryOperator randomLong) {
+    this.randomLong = randomLong;
+  }
+
+  long getRetryDelayMs() {
+    long range = MAX_RETRY_DELAY_MS - MIN_RETRY_DELAY_MS + 1;
+    return MIN_RETRY_DELAY_MS + randomLong.applyAsLong(range);
+  }
+
+  void sleep(long retryDelayMs) throws InterruptedException {
+    Thread.sleep(retryDelayMs);
+  }
+}

src/main/java/com/databricks/jdbc/api/impl/arrow/StreamingChunkDownloadTask.java

@@ -23,13 +23,13 @@ public class StreamingChunkDownloadTask implements Callable<Void> {
       JdbcLoggerFactory.getLogger(StreamingChunkDownloadTask.class);
 
   private static final int MAX_RETRIES = 5;
-  private static final long RETRY_DELAY_MS = 1500;
 
   private final ArrowResultChunk chunk;
   private final IDatabricksHttpClient httpClient;
   private final CompressionCodec compressionCodec;
   private final LinkRefresher linkRefresher;
   private final double cloudFetchSpeedThreshold;
+  private final ChunkRetryPolicy retryPolicy;
 
   // Capture caller's thread context for telemetry/logging on the download thread
   private final IDatabricksConnectionContext connectionContext;
@@ -41,11 +41,28 @@ public StreamingChunkDownloadTask(
       CompressionCodec compressionCodec,
       LinkRefresher linkRefresher,
       double cloudFetchSpeedThreshold) {
+    this(
+        chunk,
+        httpClient,
+        compressionCodec,
+        linkRefresher,
+        cloudFetchSpeedThreshold,
+        new ChunkRetryPolicy());
+  }
+
+  StreamingChunkDownloadTask(
+      ArrowResultChunk chunk,
+      IDatabricksHttpClient httpClient,
+      CompressionCodec compressionCodec,
+      LinkRefresher linkRefresher,
+      double cloudFetchSpeedThreshold,
+      ChunkRetryPolicy retryPolicy) {
     this.chunk = chunk;
     this.httpClient = httpClient;
     this.compressionCodec = compressionCodec;
     this.linkRefresher = linkRefresher;
     this.cloudFetchSpeedThreshold = cloudFetchSpeedThreshold;
+    this.retryPolicy = retryPolicy;
     this.connectionContext = DatabricksThreadContextHolder.getConnectionContext();
     this.statementId = DatabricksThreadContextHolder.getStatementId();
   }
@@ -99,11 +116,16 @@ public Void call() throws DatabricksSQLException {
                 e,
                 DatabricksDriverErrorCode.CHUNK_DOWNLOAD_ERROR);
           } else {
+            long retryDelayMs = retryPolicy.getRetryDelayMs();
             LOGGER.warn(
-                "Retry {} for chunk {}: {}", retries, chunk.getChunkIndex(), e.getMessage());
+                "Retry {} for chunk {} in {} ms: {}",
+                retries,
+                chunk.getChunkIndex(),
+                retryDelayMs,
+                e.getMessage());
             chunk.setStatus(ChunkStatus.DOWNLOAD_RETRY);
             try {
-              Thread.sleep(RETRY_DELAY_MS);
+              retryPolicy.sleep(retryDelayMs);
             } catch (InterruptedException ie) {
               Thread.currentThread().interrupt();
               throw new DatabricksSQLException(

src/test/java/com/databricks/jdbc/api/impl/arrow/ChunkDownloadTaskTest.java

@@ -39,6 +39,7 @@ public class ChunkDownloadTaskTest {
   @Mock IDatabricksHttpClient httpClient;
   @Mock RemoteChunkProvider remoteChunkProvider;
   @Mock ChunkLinkDownloadService<ArrowResultChunk> chunkLinkDownloadService;
+  @Mock ChunkRetryPolicy retryPolicy;
   private ChunkDownloadTask chunkDownloadTask;
   private CompletableFuture<Void> downloadFuture;
 
@@ -47,7 +48,8 @@ void setUp() {
     MockitoAnnotations.openMocks(this);
     downloadFuture = new CompletableFuture<>();
     chunkDownloadTask =
-        new ChunkDownloadTask(chunk, httpClient, remoteChunkProvider, chunkLinkDownloadService);
+        new ChunkDownloadTask(
+            chunk, httpClient, remoteChunkProvider, chunkLinkDownloadService, retryPolicy);
   }
 
   @Test
@@ -56,6 +58,7 @@ void testRetryLogicWithSocketException() throws Exception {
     when(chunk.isChunkLinkInvalid()).thenReturn(false);
     when(chunk.getChunkIndex()).thenReturn(7L);
     when(remoteChunkProvider.getCompressionCodec()).thenReturn(CompressionCodec.NONE);
+    when(retryPolicy.getRetryDelayMs()).thenReturn(750L, 1500L);
     DatabricksParsingException throwableError =
         new DatabricksParsingException(
             "Connection reset",
@@ -72,6 +75,9 @@ void testRetryLogicWithSocketException() throws Exception {
     chunkDownloadTask.call();
 
     verify(chunk, times(3)).downloadData(httpClient, CompressionCodec.NONE, 0.1);
+    verify(retryPolicy, times(2)).getRetryDelayMs();
+    verify(retryPolicy).sleep(750);
+    verify(retryPolicy).sleep(1500);
     assertTrue(downloadFuture.isDone());
     assertDoesNotThrow(() -> downloadFuture.get());
   }
@@ -82,6 +88,7 @@ void testRetryLogicExhaustedWithSocketException() throws Exception {
     when(chunk.isChunkLinkInvalid()).thenReturn(false);
     when(chunk.getChunkIndex()).thenReturn(7L);
     when(remoteChunkProvider.getCompressionCodec()).thenReturn(CompressionCodec.NONE);
+    when(retryPolicy.getRetryDelayMs()).thenReturn(750L, 1000L, 1250L, 1500L);
 
     // Simulate SocketException for all attempts
     doThrow(
@@ -95,6 +102,11 @@ void testRetryLogicExhaustedWithSocketException() throws Exception {
     assertThrows(DatabricksSQLException.class, () -> chunkDownloadTask.call());
     verify(chunk, times(ChunkDownloadTask.MAX_RETRIES))
         .downloadData(httpClient, CompressionCodec.NONE, 0.1);
+    verify(retryPolicy, times(ChunkDownloadTask.MAX_RETRIES - 1)).getRetryDelayMs();
+    verify(retryPolicy).sleep(750);
+    verify(retryPolicy).sleep(1000);
+    verify(retryPolicy).sleep(1250);
+    verify(retryPolicy).sleep(1500);
     assertTrue(downloadFuture.isDone());
     ExecutionException executionException =
         assertThrows(ExecutionException.class, () -> downloadFuture.get());

src/test/java/com/databricks/jdbc/api/impl/arrow/ChunkRetryPolicyTest.java

@@ -0,0 +1,28 @@
+package com.databricks.jdbc.api.impl.arrow;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.jupiter.api.Test;
+
+class ChunkRetryPolicyTest {
+
+  @Test
+  void retryDelayUsesInclusiveEqualJitterBounds() {
+    AtomicLong requestedBound = new AtomicLong();
+    ChunkRetryPolicy minimumPolicy =
+        new ChunkRetryPolicy(
+            bound -> {
+              requestedBound.set(bound);
+              return 0;
+            });
+    ChunkRetryPolicy maximumPolicy = new ChunkRetryPolicy(bound -> bound - 1);
+
+    long minimumDelay = minimumPolicy.getRetryDelayMs();
+    long maximumDelay = maximumPolicy.getRetryDelayMs();
+
+    assertEquals(751, requestedBound.get());
+    assertEquals(ChunkRetryPolicy.MIN_RETRY_DELAY_MS, minimumDelay);
+    assertEquals(ChunkRetryPolicy.MAX_RETRY_DELAY_MS, maximumDelay);
+  }
+}

src/test/java/com/databricks/jdbc/api/impl/arrow/StreamingChunkDownloadTaskTest.java

@@ -39,6 +39,7 @@ public class StreamingChunkDownloadTaskTest {
   @Mock private ArrowResultChunk chunk;
   @Mock private IDatabricksHttpClient httpClient;
   @Mock private LinkRefresher linkRefresher;
+  @Mock private ChunkRetryPolicy retryPolicy;
 
   private StreamingChunkDownloadTask downloadTask;
   private CompletableFuture<Void> downloadFuture;
@@ -48,7 +49,12 @@ void setUp() {
     downloadFuture = new CompletableFuture<>();
     downloadTask =
         new StreamingChunkDownloadTask(
-            chunk, httpClient, CompressionCodec.NONE, linkRefresher, CLOUD_FETCH_SPEED_THRESHOLD);
+            chunk,
+            httpClient,
+            CompressionCodec.NONE,
+            linkRefresher,
+            CLOUD_FETCH_SPEED_THRESHOLD,
+            retryPolicy);
   }
 
   @Test
@@ -67,6 +73,7 @@ void testSuccessfulDownloadOnFirstAttempt() throws Exception {
     verify(chunk, times(1))
         .downloadData(httpClient, CompressionCodec.NONE, CLOUD_FETCH_SPEED_THRESHOLD);
     verify(chunk, never()).setStatus(ChunkStatus.DOWNLOAD_RETRY);
+    verifyNoInteractions(retryPolicy);
     assertTrue(downloadFuture.isDone());
     assertDoesNotThrow(() -> downloadFuture.get());
   }
@@ -76,6 +83,7 @@ void testRetryLogicWithSocketException() throws Exception {
     when(chunk.getChunkReadyFuture()).thenReturn(downloadFuture);
     when(chunk.isChunkLinkInvalid()).thenReturn(false);
     when(chunk.getChunkIndex()).thenReturn(7L);
+    when(retryPolicy.getRetryDelayMs()).thenReturn(750L, 1500L);
 
     DatabricksParsingException throwableError =
         new DatabricksParsingException(
@@ -95,6 +103,9 @@ void testRetryLogicWithSocketException() throws Exception {
     verify(chunk, times(3))
         .downloadData(httpClient, CompressionCodec.NONE, CLOUD_FETCH_SPEED_THRESHOLD);
     verify(chunk, times(2)).setStatus(ChunkStatus.DOWNLOAD_RETRY);
+    verify(retryPolicy, times(2)).getRetryDelayMs();
+    verify(retryPolicy).sleep(750);
+    verify(retryPolicy).sleep(1500);
     assertTrue(downloadFuture.isDone());
     assertDoesNotThrow(() -> downloadFuture.get());
   }
@@ -104,6 +115,7 @@ void testRetryLogicExhaustedWithSocketException() throws Exception {
     when(chunk.getChunkReadyFuture()).thenReturn(downloadFuture);
     when(chunk.isChunkLinkInvalid()).thenReturn(false);
     when(chunk.getChunkIndex()).thenReturn(7L);
+    when(retryPolicy.getRetryDelayMs()).thenReturn(750L, 1000L, 1250L, 1500L);
 
     // Simulate SocketException for all attempts
     doThrow(
@@ -120,6 +132,11 @@ void testRetryLogicExhaustedWithSocketException() throws Exception {
     verify(chunk, times(5))
         .downloadData(httpClient, CompressionCodec.NONE, CLOUD_FETCH_SPEED_THRESHOLD);
     verify(chunk, times(1)).setStatus(ChunkStatus.DOWNLOAD_FAILED);
+    verify(retryPolicy, times(4)).getRetryDelayMs();
+    verify(retryPolicy).sleep(750);
+    verify(retryPolicy).sleep(1000);
+    verify(retryPolicy).sleep(1250);
+    verify(retryPolicy).sleep(1500);
     assertTrue(downloadFuture.isDone());
 
     ExecutionException executionException =