diff --git a/NEXT_CHANGELOG.md b/NEXT_CHANGELOG.md index e12c8875f..e956b1b17 100644 --- a/NEXT_CHANGELOG.md +++ b/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. diff --git a/src/main/java/com/databricks/jdbc/api/impl/arrow/ChunkDownloadTask.java b/src/main/java/com/databricks/jdbc/api/impl/arrow/ChunkDownloadTask.java index 217b74a21..9fca6dcf8 100644 --- a/src/main/java/com/databricks/jdbc/api/impl/arrow/ChunkDownloadTask.java +++ b/src/main/java/com/databricks/jdbc/api/impl/arrow/ChunkDownloadTask.java @@ -17,13 +17,13 @@ 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 linkDownloadService; + private final ChunkRetryPolicy retryPolicy; Throwable uncaughtException = null; ChunkDownloadTask( @@ -31,12 +31,22 @@ class ChunkDownloadTask implements DatabricksCallableTask { IDatabricksHttpClient httpClient, ChunkDownloadManager chunkDownloader, ChunkLinkDownloadService linkDownloadService) { + this(chunk, httpClient, chunkDownloader, linkDownloadService, new ChunkRetryPolicy()); + } + + ChunkDownloadTask( + ArrowResultChunk chunk, + IDatabricksHttpClient httpClient, + ChunkDownloadManager chunkDownloader, + ChunkLinkDownloadService 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( diff --git a/src/main/java/com/databricks/jdbc/api/impl/arrow/ChunkRetryPolicy.java b/src/main/java/com/databricks/jdbc/api/impl/arrow/ChunkRetryPolicy.java new file mode 100644 index 000000000..ddecf73ef --- /dev/null +++ b/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); + } +} diff --git a/src/main/java/com/databricks/jdbc/api/impl/arrow/StreamingChunkDownloadTask.java b/src/main/java/com/databricks/jdbc/api/impl/arrow/StreamingChunkDownloadTask.java index 177d0bf9f..c318fb410 100644 --- a/src/main/java/com/databricks/jdbc/api/impl/arrow/StreamingChunkDownloadTask.java +++ b/src/main/java/com/databricks/jdbc/api/impl/arrow/StreamingChunkDownloadTask.java @@ -23,13 +23,13 @@ public class StreamingChunkDownloadTask implements Callable { 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( diff --git a/src/test/java/com/databricks/jdbc/api/impl/arrow/ChunkDownloadTaskTest.java b/src/test/java/com/databricks/jdbc/api/impl/arrow/ChunkDownloadTaskTest.java index b10336f1e..5d64c45b3 100644 --- a/src/test/java/com/databricks/jdbc/api/impl/arrow/ChunkDownloadTaskTest.java +++ b/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 chunkLinkDownloadService; + @Mock ChunkRetryPolicy retryPolicy; private ChunkDownloadTask chunkDownloadTask; private CompletableFuture 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()); diff --git a/src/test/java/com/databricks/jdbc/api/impl/arrow/ChunkRetryPolicyTest.java b/src/test/java/com/databricks/jdbc/api/impl/arrow/ChunkRetryPolicyTest.java new file mode 100644 index 000000000..3165c7289 --- /dev/null +++ b/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); + } +} diff --git a/src/test/java/com/databricks/jdbc/api/impl/arrow/StreamingChunkDownloadTaskTest.java b/src/test/java/com/databricks/jdbc/api/impl/arrow/StreamingChunkDownloadTaskTest.java index 4d2e5aed1..824f0d7c0 100644 --- a/src/test/java/com/databricks/jdbc/api/impl/arrow/StreamingChunkDownloadTaskTest.java +++ b/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 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 =