Added new connection property

src/main/java/net/snowflake/client/core/SFBaseSession.java

@@ -146,6 +146,8 @@ public abstract class SFBaseSession {
 
   private boolean isJdbcArrowTreatDecimalAsInt = true;
 
+  private boolean supportImplicitAsyncQueryTimeout = false;
+
   protected SFBaseSession(SFConnectionHandler sfConnectionHandler) {
     this.sfConnectionHandler = sfConnectionHandler;
   }
@@ -1314,4 +1316,20 @@ public SFConnectionHandler getSfConnectionHandler() {
   public boolean getEnableReturnTimestampWithTimeZone() {
     return enableReturnTimestampWithTimeZone;
   }
+
+  /**
+   * @return True if query timeout should be set on the server side for async queries. False by
+   *     default.
+   */
+  public boolean getSupportImplicitAsyncQueryTimeout() {
+    return supportImplicitAsyncQueryTimeout;
+  }
+
+  /**
+   * @param supportImplicitAsyncQueryTimeout Setting supportImplicitAsyncQueryTimeout to true allows
+   *     for query timeout to be set on the server side.
+   */
+  public void setSupportImplicitAsyncQueryTimeout(boolean supportImplicitAsyncQueryTimeout) {
+    this.supportImplicitAsyncQueryTimeout = supportImplicitAsyncQueryTimeout;
+  }
 }

src/main/java/net/snowflake/client/core/SFBaseStatement.java

@@ -42,8 +42,10 @@ public void addProperty(String propertyName, Object propertyValue) throws SFExce
     if ("query_timeout".equalsIgnoreCase(propertyName)) {
       // Client side implementation
       queryTimeout = (Integer) propertyValue;
-      // Set server parameter for supporting query timeout on async queries
-      statementParametersMap.put("STATEMENT_TIMEOUT_IN_SECONDS", (Integer) propertyValue);
+      if (this.getSFBaseSession().getSupportImplicitAsyncQueryTimeout()) {
+        // Set server parameter for supporting query timeout on async queries
+        statementParametersMap.put("STATEMENT_TIMEOUT_IN_SECONDS", (Integer) propertyValue);
+      }
     }
 
     // check if the number of session properties exceed limit

src/main/java/net/snowflake/client/core/SFSession.java

@@ -537,6 +537,12 @@ public void addSFSessionProperty(String propertyName, Object propertyValue) thro
           }
           break;
 
+        case SUPPORT_IMPLICIT_ASYNC_QUERY_TIMEOUT:
+          if (propertyValue != null) {
+            setSupportImplicitAsyncQueryTimeout(getBooleanValue(propertyValue));
+          }
+          break;
+
         default:
           break;
       }

src/main/java/net/snowflake/client/core/SFSessionProperty.java

@@ -112,7 +112,10 @@ public enum SFSessionProperty {
 
   HTTP_CLIENT_CONNECTION_TIMEOUT("HTTP_CLIENT_CONNECTION_TIMEOUT", false, Integer.class),
 
-  HTTP_CLIENT_SOCKET_TIMEOUT("HTTP_CLIENT_SOCKET_TIMEOUT", false, Integer.class);
+  HTTP_CLIENT_SOCKET_TIMEOUT("HTTP_CLIENT_SOCKET_TIMEOUT", false, Integer.class),
+
+  SUPPORT_IMPLICIT_ASYNC_QUERY_TIMEOUT(
+      "SUPPORT_IMPLICIT_ASYNC_QUERY_TIMEOUT", false, Boolean.class);
 
   // property key in string
   private String propertyKey;

src/test/java/net/snowflake/client/jdbc/StatementLatestIT.java

@@ -4,6 +4,7 @@
 package net.snowflake.client.jdbc;
 
 import static net.snowflake.client.jdbc.ErrorCode.ROW_DOES_NOT_EXIST;
+import static org.awaitility.Awaitility.await;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -19,9 +20,11 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import net.snowflake.client.TestUtil;
 import net.snowflake.client.annotations.DontRunOnGithubActions;
 import net.snowflake.client.category.TestTags;
@@ -305,33 +308,26 @@ public void testQueryIdIsSetOnFailedExecuteQuery() throws SQLException {
    */
   @Test
   public void testSetQueryTimeoutForAsnycQuery() throws SQLException {
-    try (Statement statement = connection.createStatement()) {
-      statement.unwrap(SnowflakeStatement.class).setParameter("MULTI_STATEMENT_COUNT", 0);
+    Properties p = new Properties();
+    p.put("SUPPORT_IMPLICIT_ASYNC_QUERY_TIMEOUT", true);
+    try (Connection con = getConnection(p);
+        Statement statement = con.createStatement()) {
       statement.setQueryTimeout(3);
 
       String sql = "SELECT * FROM SNOWFLAKE_SAMPLE_DATA.TPCDS_SF100TCL.CUSTOMER;";
 
       try (ResultSet resultSet =
           statement.unwrap(SnowflakeStatement.class).executeAsyncQuery(sql)) {
-
-        QueryStatus queryStatus = QueryStatus.RUNNING;
-        while (queryStatus == QueryStatus.RUNNING
-            || queryStatus == QueryStatus.RESUMING_WAREHOUSE) {
-          try {
-            Thread.sleep(100);
-          } catch (InterruptedException e) {
-            fail(e.getMessage());
-          }
-          queryStatus = resultSet.unwrap(SnowflakeResultSet.class).getStatus();
-        }
-
-        if (queryStatus == QueryStatus.FAILED_WITH_ERROR) {
-          assertTrue(
-              queryStatus
-                  .getErrorMessage()
-                  .contains(
-                      "Statement reached its statement or warehouse timeout of 3 second(s) and was canceled"));
-        }
+        SnowflakeResultSet sfrs = resultSet.unwrap(SnowflakeResultSet.class);
+        await()
+            .atMost(Duration.ofSeconds(5))
+            .until(() -> sfrs.getStatusV2().getStatus() == QueryStatus.FAILED_WITH_ERROR);
+
+        assertTrue(
+            sfrs.getStatusV2()
+                .getErrorMessage()
+                .contains(
+                    "Statement reached its statement or warehouse timeout of 3 second(s) and was canceled"));
       }
     }
   }