Merge branch 'master' into azagrebin-SNOW-835618-send-uncompressed-ch…

…unk-len

src/main/java/net/snowflake/ingest/streaming/internal/StreamingIngestUtils.java

@@ -34,9 +34,40 @@ public Long apply(T input) {
 
   private static final ObjectMapper objectMapper = new ObjectMapper();
 
-  static void sleepForRetry(int executionCount) {
+  /**
+   * How many milliseconds of exponential backoff to sleep before retrying the request again:
+   *
+   * <ul>
+   *   <li>0 or 1 failure => no sleep
+   *   <li>2 failures => 1s
+   *   <li>3 failures => 2s
+   *   <li>4 or more failures => 4s
+   * </ul>
+   *
+   * @param executionCount How many unsuccessful attempts have been attempted
+   * @return Sleep time in ms
+   */
+  static long getSleepForRetryMs(int executionCount) {
+    if (executionCount < 0) {
+      throw new IllegalArgumentException(
+          String.format(
+              "executionCount must be a non-negative integer, passed: %d", executionCount));
+    } else if (executionCount < 2) {
+      return 0;
+    } else {
+      final int effectiveExecutionCount = Math.min(executionCount, 4);
+      return (1 << (effectiveExecutionCount - 2)) * 1000L;
+    }
+  }
+
+  public static void sleepForRetry(int executionCount) {
+    long sleepForRetryMs = getSleepForRetryMs(executionCount);
+    if (sleepForRetryMs == 0) {
+      return;
+    }
+
     try {
-      Thread.sleep((1 << (executionCount + 1)) * 1000);
+      Thread.sleep(sleepForRetryMs);
     } catch (InterruptedException e) {
       throw new SFException(ErrorCode.INTERNAL_ERROR, e.getMessage());
     }

src/main/java/net/snowflake/ingest/utils/HttpUtil.java

@@ -40,6 +40,7 @@
 import net.snowflake.client.jdbc.internal.apache.http.pool.PoolStats;
 import net.snowflake.client.jdbc.internal.apache.http.protocol.HttpContext;
 import net.snowflake.client.jdbc.internal.apache.http.ssl.SSLContexts;
+import net.snowflake.ingest.streaming.internal.StreamingIngestUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -57,7 +58,14 @@ public class HttpUtil {
   private static final String FIRST_FAULT_TIMESTAMP = "FIRST_FAULT_TIMESTAMP";
   private static final Duration TOTAL_RETRY_DURATION = Duration.of(120, ChronoUnit.SECONDS);
   private static final Duration RETRY_INTERVAL = Duration.of(3, ChronoUnit.SECONDS);
-  private static final int MAX_RETRIES = 3;
+
+  /**
+   * How many times to retry when an IO exception is thrown. Value here is chosen to match the total
+   * value of {@link HttpUtil.TOTAL_RETRY_DURATION} when exponential backoff of up to 4 seconds per
+   * retry is used.
+   */
+  private static final int MAX_RETRIES = 10;
+
   private static volatile CloseableHttpClient httpClient;
 
   private static PoolingHttpClientConnectionManager connectionManager;
@@ -284,13 +292,16 @@ static HttpRequestRetryHandler getHttpRequestRetryHandler() {
         return false;
       }
       if (exception instanceof NoHttpResponseException
-          || exception instanceof javax.net.ssl.SSLException) {
+          || exception instanceof javax.net.ssl.SSLException
+          || exception instanceof java.net.SocketException
+          || exception instanceof java.net.UnknownHostException) {
         LOGGER.info(
             "Retrying request which caused {} with " + "URI:{}, retryCount:{} and maxRetryCount:{}",
             exception.getClass().getName(),
             requestURI,
             executionCount,
             MAX_RETRIES);
+        StreamingIngestUtils.sleepForRetry(executionCount);
         return true;
       }
       LOGGER.info("No retry for URI:{} with exception {}", requestURI, exception.toString());

src/test/java/net/snowflake/ingest/streaming/internal/StreamingIngestUtilsTest.java

@@ -2,6 +2,7 @@
 
 import static net.snowflake.ingest.connection.ServiceResponseHandler.ApiName.STREAMING_CHANNEL_STATUS;
 import static net.snowflake.ingest.streaming.internal.StreamingIngestUtils.executeWithRetries;
+import static net.snowflake.ingest.streaming.internal.StreamingIngestUtils.getSleepForRetryMs;
 import static net.snowflake.ingest.utils.Constants.CHANNEL_STATUS_ENDPOINT;
 import static net.snowflake.ingest.utils.Constants.RESPONSE_ERR_GENERAL_EXCEPTION_RETRY_REQUEST;
 import static net.snowflake.ingest.utils.Constants.RESPONSE_SUCCESS;
@@ -162,4 +163,20 @@ public void testRetriesRecovery() throws Exception {
 
     Assert.assertEquals("honkSuccess", result.getMessage());
   }
+
+  @Test
+  public void testGetSleepForRetry() {
+    Assert.assertEquals(0, getSleepForRetryMs(0));
+    Assert.assertEquals(0, getSleepForRetryMs(1));
+    Assert.assertEquals(1000, getSleepForRetryMs(2));
+    Assert.assertEquals(2000, getSleepForRetryMs(3));
+    Assert.assertEquals(4000, getSleepForRetryMs(4));
+    Assert.assertEquals(4000, getSleepForRetryMs(5));
+    Assert.assertEquals(4000, getSleepForRetryMs(100000));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testGetSleepForRetryNegative() {
+    getSleepForRetryMs(-1);
+  }
 }

src/test/java/net/snowflake/ingest/utils/HttpUtilTest.java

@@ -5,6 +5,8 @@
 import static org.mockito.Mockito.doReturn;
 
 import java.io.IOException;
+import java.net.SocketException;
+import java.net.UnknownHostException;
 import javax.net.ssl.SSLException;
 import net.snowflake.client.jdbc.internal.apache.http.HttpRequest;
 import net.snowflake.client.jdbc.internal.apache.http.NoHttpResponseException;
@@ -33,9 +35,15 @@ public void testRequestRetryHandler() {
     assertTrue(
         httpRequestRetryHandler.retryRequest(
             new SSLException("Test exception"), 1, httpContextMock));
+    assertTrue(
+        httpRequestRetryHandler.retryRequest(
+            new SocketException("Test exception"), 1, httpContextMock));
+    assertTrue(
+        httpRequestRetryHandler.retryRequest(
+            new UnknownHostException("Test exception"), 1, httpContextMock));
     assertFalse(
         httpRequestRetryHandler.retryRequest(
-            new SSLException("Test exception"), 4, httpContextMock));
+            new SSLException("Test exception"), 11, httpContextMock));
     assertFalse(httpRequestRetryHandler.retryRequest(new IOException(), 1, httpContextMock));
   }
 }