diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryBaseService.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryBaseService.java
index 976015873..bdcefb3d9 100644
--- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryBaseService.java
+++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryBaseService.java
@@ -26,7 +26,7 @@ protected BigQueryBaseService(ServiceOptions options) {
     super(options);
   }
 
-  public static final ExceptionHandler BIGQUERY_EXCEPTION_HANDLER =
+  public static final ExceptionHandler DEFAULT_BIGQUERY_EXCEPTION_HANDLER =
       ExceptionHandler.newBuilder()
           .abortOn(RuntimeException.class)
           .retryOn(java.net.ConnectException.class) // retry on Connection Exception
diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryImpl.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryImpl.java
index a32a600c3..088d15c09 100644
--- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryImpl.java
+++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryImpl.java
@@ -293,7 +293,7 @@ public com.google.api.services.bigquery.model.Dataset call() throws IOException
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -340,7 +340,7 @@ public com.google.api.services.bigquery.model.Table call() throws IOException {
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -396,7 +396,7 @@ public com.google.api.services.bigquery.model.Routine call() throws IOException
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -490,7 +490,7 @@ public com.google.api.services.bigquery.model.Job call() throws IOException {
                       ? RetryOption.mergeToSettings(
                           getOptions().getRetrySettings(), getRetryOptions(optionsMap))
                       : getOptions().getRetrySettings(),
-                  BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+                  getOptions().getResultRetryAlgorithm(),
                   getOptions().getClock(),
                   getBigQueryRetryConfig(optionsMap) != null
                       ? getBigQueryRetryConfig(optionsMap)
@@ -586,7 +586,7 @@ public com.google.api.services.bigquery.model.Dataset call() throws IOException
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -652,7 +652,7 @@ private static Page<Dataset> listDatasets(
                 }
               },
               serviceOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              serviceOptions.getResultRetryAlgorithm(),
               serviceOptions.getClock(),
               EMPTY_RETRY_CONFIG,
               serviceOptions.isOpenTelemetryTracingEnabled(),
@@ -704,7 +704,7 @@ public Boolean call() throws IOException {
             }
           },
           getOptions().getRetrySettings(),
-          BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+          getOptions().getResultRetryAlgorithm(),
           getOptions().getClock(),
           EMPTY_RETRY_CONFIG,
           getOptions().isOpenTelemetryTracingEnabled(),
@@ -755,7 +755,7 @@ public Boolean call() throws IOException {
             }
           },
           getOptions().getRetrySettings(),
-          BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+          getOptions().getResultRetryAlgorithm(),
           getOptions().getClock(),
           EMPTY_RETRY_CONFIG,
           getOptions().isOpenTelemetryTracingEnabled(),
@@ -801,7 +801,7 @@ public Boolean call() throws IOException {
             }
           },
           getOptions().getRetrySettings(),
-          BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+          getOptions().getResultRetryAlgorithm(),
           getOptions().getClock(),
           EMPTY_RETRY_CONFIG,
           getOptions().isOpenTelemetryTracingEnabled(),
@@ -847,7 +847,7 @@ public Boolean call() throws IOException {
             }
           },
           getOptions().getRetrySettings(),
-          BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+          getOptions().getResultRetryAlgorithm(),
           getOptions().getClock(),
           EMPTY_RETRY_CONFIG,
           getOptions().isOpenTelemetryTracingEnabled(),
@@ -891,7 +891,7 @@ public Boolean call() throws IOException {
             }
           },
           getOptions().getRetrySettings(),
-          BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+          getOptions().getResultRetryAlgorithm(),
           getOptions().getClock(),
           EMPTY_RETRY_CONFIG,
           getOptions().isOpenTelemetryTracingEnabled(),
@@ -932,7 +932,7 @@ public com.google.api.services.bigquery.model.Dataset call() throws IOException
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -979,7 +979,7 @@ public com.google.api.services.bigquery.model.Table call() throws IOException {
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -1025,7 +1025,7 @@ public com.google.api.services.bigquery.model.Model call() throws IOException {
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -1071,7 +1071,7 @@ public com.google.api.services.bigquery.model.Routine call() throws IOException
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -1125,7 +1125,7 @@ public com.google.api.services.bigquery.model.Table call() throws IOException {
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -1184,7 +1184,7 @@ public com.google.api.services.bigquery.model.Model call() throws IOException {
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -1243,7 +1243,7 @@ public com.google.api.services.bigquery.model.Routine call() throws IOException
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -1461,7 +1461,7 @@ public Tuple<String, Iterable<com.google.api.services.bigquery.model.Table>> cal
                 }
               },
               serviceOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              serviceOptions.getResultRetryAlgorithm(),
               serviceOptions.getClock(),
               EMPTY_RETRY_CONFIG,
               serviceOptions.isOpenTelemetryTracingEnabled(),
@@ -1502,7 +1502,7 @@ public Tuple<String, Iterable<com.google.api.services.bigquery.model.Model>> cal
                 }
               },
               serviceOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              serviceOptions.getResultRetryAlgorithm(),
               serviceOptions.getClock(),
               EMPTY_RETRY_CONFIG,
               serviceOptions.isOpenTelemetryTracingEnabled(),
@@ -1543,7 +1543,7 @@ private static Page<Routine> listRoutines(
                 }
               },
               serviceOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              serviceOptions.getResultRetryAlgorithm(),
               serviceOptions.getClock(),
               EMPTY_RETRY_CONFIG,
               serviceOptions.isOpenTelemetryTracingEnabled(),
@@ -1625,7 +1625,7 @@ public TableDataInsertAllResponse call() throws Exception {
                     }
                   },
                   getOptions().getRetrySettings(),
-                  BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+                  getOptions().getResultRetryAlgorithm(),
                   getOptions().getClock(),
                   EMPTY_RETRY_CONFIG,
                   getOptions().isOpenTelemetryTracingEnabled(),
@@ -1719,7 +1719,7 @@ public TableDataList call() throws IOException {
                 }
               },
               serviceOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              serviceOptions.getResultRetryAlgorithm(),
               serviceOptions.getClock(),
               EMPTY_RETRY_CONFIG,
               serviceOptions.isOpenTelemetryTracingEnabled(),
@@ -1793,7 +1793,7 @@ public com.google.api.services.bigquery.model.Job call() throws IOException {
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -1850,7 +1850,7 @@ public Tuple<String, Iterable<com.google.api.services.bigquery.model.Job>> call(
                 }
               },
               serviceOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              serviceOptions.getResultRetryAlgorithm(),
               serviceOptions.getClock(),
               EMPTY_RETRY_CONFIG,
               serviceOptions.isOpenTelemetryTracingEnabled(),
@@ -1905,7 +1905,7 @@ public Boolean call() throws IOException {
             }
           },
           getOptions().getRetrySettings(),
-          BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+          getOptions().getResultRetryAlgorithm(),
           getOptions().getClock(),
           EMPTY_RETRY_CONFIG,
           getOptions().isOpenTelemetryTracingEnabled(),
@@ -1992,7 +1992,7 @@ public com.google.api.services.bigquery.model.QueryResponse call()
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               DEFAULT_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -2169,7 +2169,7 @@ public GetQueryResultsResponse call() throws IOException {
                 }
               },
               serviceOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              serviceOptions.getResultRetryAlgorithm(),
               serviceOptions.getClock(),
               DEFAULT_RETRY_CONFIG,
               serviceOptions.isOpenTelemetryTracingEnabled(),
@@ -2240,7 +2240,7 @@ public com.google.api.services.bigquery.model.Policy call() throws IOException {
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -2286,7 +2286,7 @@ public com.google.api.services.bigquery.model.Policy call() throws IOException {
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -2334,7 +2334,7 @@ public com.google.api.services.bigquery.model.TestIamPermissionsResponse call()
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryOptions.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryOptions.java
index 36eeb2086..7adb58d3a 100644
--- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryOptions.java
+++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryOptions.java
@@ -17,6 +17,7 @@
 package com.google.cloud.bigquery;
 
 import com.google.api.core.BetaApi;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.cloud.ServiceDefaults;
 import com.google.cloud.ServiceOptions;
 import com.google.cloud.ServiceRpc;
@@ -43,6 +44,7 @@ public class BigQueryOptions extends ServiceOptions<BigQuery, BigQueryOptions> {
   private JobCreationMode defaultJobCreationMode = JobCreationMode.JOB_CREATION_MODE_UNSPECIFIED;
   private boolean enableOpenTelemetryTracing;
   private Tracer openTelemetryTracer;
+  private ResultRetryAlgorithm<?> resultRetryAlgorithm;
 
   public static class DefaultBigQueryFactory implements BigQueryFactory {
 
@@ -70,6 +72,7 @@ public static class Builder extends ServiceOptions.Builder<BigQuery, BigQueryOpt
     private boolean useInt64Timestamps;
     private boolean enableOpenTelemetryTracing;
     private Tracer openTelemetryTracer;
+    private ResultRetryAlgorithm<?> resultRetryAlgorithm;
 
     private Builder() {}
 
@@ -118,6 +121,11 @@ public Builder setOpenTelemetryTracer(Tracer tracer) {
       return this;
     }
 
+    public Builder setResultRetryAlgorithm(ResultRetryAlgorithm<?> resultRetryAlgorithm) {
+      this.resultRetryAlgorithm = resultRetryAlgorithm;
+      return this;
+    }
+
     @Override
     public BigQueryOptions build() {
       return new BigQueryOptions(this);
@@ -130,6 +138,11 @@ private BigQueryOptions(Builder builder) {
     this.useInt64Timestamps = builder.useInt64Timestamps;
     this.enableOpenTelemetryTracing = builder.enableOpenTelemetryTracing;
     this.openTelemetryTracer = builder.openTelemetryTracer;
+    if (builder.resultRetryAlgorithm != null) {
+      this.resultRetryAlgorithm = builder.resultRetryAlgorithm;
+    } else {
+      this.resultRetryAlgorithm = BigQueryBaseService.DEFAULT_BIGQUERY_EXCEPTION_HANDLER;
+    }
   }
 
   private static class BigQueryDefaults implements ServiceDefaults<BigQuery, BigQueryOptions> {
@@ -221,6 +234,10 @@ public Tracer getOpenTelemetryTracer() {
     return openTelemetryTracer;
   }
 
+  public ResultRetryAlgorithm<?> getResultRetryAlgorithm() {
+    return resultRetryAlgorithm;
+  }
+
   @SuppressWarnings("unchecked")
   @Override
   public Builder toBuilder() {
diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java
index b12ad8556..c3465c33a 100644
--- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java
+++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java
@@ -474,7 +474,7 @@ private BigQueryResult queryRpc(
                       .getBigQueryRpcV2()
                       .queryRpcSkipExceptionTranslation(projectId, queryRequest),
               bigQueryOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              bigQueryOptions.getResultRetryAlgorithm(),
               bigQueryOptions.getClock(),
               retryConfig,
               false,
@@ -932,7 +932,7 @@ private Job getQueryJobRpc(JobId jobId) {
                           completeJobId.getJob(),
                           completeJobId.getLocation()),
               bigQueryOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              bigQueryOptions.getResultRetryAlgorithm(),
               bigQueryOptions.getClock(),
               EMPTY_RETRY_CONFIG,
               false,
@@ -979,7 +979,7 @@ TableDataList tableDataListRpc(TableId destinationTable, String pageToken) {
                           connectionSettings.getMaxResultPerPage(),
                           pageToken),
               bigQueryOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              bigQueryOptions.getResultRetryAlgorithm(),
               bigQueryOptions.getClock(),
               EMPTY_RETRY_CONFIG,
               false,
@@ -1212,7 +1212,7 @@ GetQueryResultsResponse getQueryResultsFirstPage(JobId jobId) {
                             connectionSettings.getMaxResultPerPage(),
                             timeoutMs),
                 bigQueryOptions.getRetrySettings(),
-                BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+                bigQueryOptions.getResultRetryAlgorithm(),
                 bigQueryOptions.getClock(),
                 retryConfig,
                 false,
@@ -1477,7 +1477,7 @@ com.google.api.services.bigquery.model.Job createQueryJob(
                       .getBigQueryRpcV2()
                       .createJobForQuerySkipExceptionTranslation(jobPb),
               bigQueryOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              bigQueryOptions.getResultRetryAlgorithm(),
               bigQueryOptions.getClock(),
               retryConfig,
               false,
@@ -1522,7 +1522,7 @@ com.google.api.services.bigquery.model.Job createDryRunJob(String sql) {
                       .getBigQueryRpcV2()
                       .createJobForQuerySkipExceptionTranslation(jobPb),
               bigQueryOptions.getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              bigQueryOptions.getResultRetryAlgorithm(),
               bigQueryOptions.getClock(),
               retryConfig,
               false,
diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/TableDataWriteChannel.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/TableDataWriteChannel.java
index cc64ec72f..dde4c0818 100644
--- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/TableDataWriteChannel.java
+++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/TableDataWriteChannel.java
@@ -77,7 +77,7 @@ public com.google.api.services.bigquery.model.Job call() throws IOException {
                 }
               },
               getOptions().getRetrySettings(),
-              BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+              getOptions().getResultRetryAlgorithm(),
               getOptions().getClock(),
               EMPTY_RETRY_CONFIG,
               getOptions().isOpenTelemetryTracingEnabled(),
@@ -126,7 +126,7 @@ public String call() throws IOException {
             }
           },
           options.getRetrySettings(),
-          BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER,
+          options.getResultRetryAlgorithm(),
           options.getClock(),
           EMPTY_RETRY_CONFIG,
           options.isOpenTelemetryTracingEnabled(),
diff --git a/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/BigQueryExceptionTest.java b/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/BigQueryExceptionTest.java
index 41915e283..8a2094b55 100644
--- a/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/BigQueryExceptionTest.java
+++ b/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/BigQueryExceptionTest.java
@@ -21,7 +21,10 @@
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -29,9 +32,14 @@
 import com.google.api.client.http.HttpHeaders;
 import com.google.api.client.http.HttpResponseException;
 import com.google.cloud.BaseServiceException;
+import com.google.cloud.ExceptionHandler;
 import com.google.cloud.RetryHelper.RetryHelperException;
+import com.google.cloud.bigquery.spi.v2.BigQueryRpc;
+import com.google.cloud.bigquery.spi.v2.HttpBigQueryRpc;
 import java.io.IOException;
 import java.net.SocketTimeoutException;
+import java.util.HashMap;
+import java.util.Map;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.junit.MockitoJUnitRunner;
@@ -158,4 +166,87 @@ public void testTranslateAndThrow() throws Exception {
       verify(exceptionMock, times(2)).getCause();
     }
   }
+
+  @Test
+  public void testDefaultExceptionHandler() throws java.io.IOException {
+    BigQueryOptions defaultOptions =
+        BigQueryOptions.newBuilder().setProjectId("project-id").build();
+    DatasetInfo info = DatasetInfo.newBuilder("dataset").build();
+    Dataset dataset = null;
+
+    final com.google.api.services.bigquery.model.Dataset datasetPb =
+        info.setProjectId(defaultOptions.getProjectId()).toPb();
+    final Map<BigQueryRpc.Option, ?> optionsMap = new HashMap<>();
+
+    BigQueryOptions mockOptions = spy(defaultOptions);
+    HttpBigQueryRpc bigQueryRpcMock = mock(HttpBigQueryRpc.class);
+    doReturn(bigQueryRpcMock).when(mockOptions).getBigQueryRpcV2();
+    // java.net.SocketException is retry-able in the default exception handler.
+    doThrow(java.net.SocketException.class)
+        .when(bigQueryRpcMock)
+        .createSkipExceptionTranslation(datasetPb, optionsMap);
+
+    BigQuery bigquery = mockOptions.getService();
+    try {
+      dataset = bigquery.create(info);
+    } catch (BigQueryException e) {
+      assertEquals(e.getCause().getClass(), java.net.SocketException.class);
+      assertNull(dataset);
+    } finally {
+      verify(bigQueryRpcMock, times(6)).createSkipExceptionTranslation(datasetPb, optionsMap);
+    }
+  }
+
+  @Test
+  public void testCustomExceptionHandler() throws java.io.IOException {
+    BigQueryOptions defaultOptions =
+        BigQueryOptions.newBuilder()
+            .setProjectId("project-id")
+            .setResultRetryAlgorithm(
+                ExceptionHandler.newBuilder()
+                    .abortOn(RuntimeException.class)
+                    .retryOn(java.util.EmptyStackException.class)
+                    .addInterceptors(BigQueryBaseService.EXCEPTION_HANDLER_INTERCEPTOR)
+                    .build())
+            .build();
+    DatasetInfo info = DatasetInfo.newBuilder("dataset").build();
+    Dataset dataset = null;
+
+    final com.google.api.services.bigquery.model.Dataset datasetPb =
+        info.setProjectId(defaultOptions.getProjectId()).toPb();
+    final Map<BigQueryRpc.Option, ?> optionsMap = new HashMap<>();
+
+    BigQueryOptions mockOptions = spy(defaultOptions);
+    HttpBigQueryRpc bigQueryRpcRetryMock = mock(HttpBigQueryRpc.class);
+    doReturn(bigQueryRpcRetryMock).when(mockOptions).getBigQueryRpcV2();
+    doThrow(java.util.EmptyStackException.class)
+        .when(bigQueryRpcRetryMock)
+        .createSkipExceptionTranslation(datasetPb, optionsMap);
+
+    BigQuery bigquery = mockOptions.getService();
+    try {
+      dataset = bigquery.create(info);
+    } catch (BigQueryException e) {
+      assertEquals(e.getCause().getClass(), java.util.EmptyStackException.class);
+      assertNull(dataset);
+    } finally {
+      verify(bigQueryRpcRetryMock, times(6)).createSkipExceptionTranslation(datasetPb, optionsMap);
+    }
+
+    BigQueryOptions mockOptionsAbort = spy(defaultOptions);
+    HttpBigQueryRpc bigQueryRpcAbortMock = mock(HttpBigQueryRpc.class);
+    doReturn(bigQueryRpcAbortMock).when(mockOptionsAbort).getBigQueryRpcV2();
+    doThrow(RuntimeException.class)
+        .when(bigQueryRpcAbortMock)
+        .createSkipExceptionTranslation(datasetPb, optionsMap);
+    bigquery = mockOptionsAbort.getService();
+    try {
+      dataset = bigquery.create(info);
+    } catch (BigQueryException e) {
+      assertEquals(e.getCause().getClass(), RuntimeException.class);
+      assertNull(dataset);
+    } finally {
+      verify(bigQueryRpcAbortMock, times(1)).createSkipExceptionTranslation(datasetPb, optionsMap);
+    }
+  }
 }