[JDBC] NetworkTimeout

client-v2/src/main/java/com/clickhouse/client/api/insert/InsertSettings.java

@@ -5,6 +5,7 @@
 import com.clickhouse.client.api.ClientConfigProperties;
 import com.clickhouse.client.api.enums.Protocol;
 import com.clickhouse.client.api.internal.ValidationUtils;
+import com.clickhouse.client.api.query.QuerySettings;
 import org.apache.hc.core5.http.HttpHeaders;
 
 import java.util.Collection;
@@ -264,4 +265,27 @@ public InsertSettings logComment(String logComment) {
     public String getLogComment() {
         return logComment;
     }
+
+    public static InsertSettings merge(InsertSettings source, InsertSettings override) {
+        InsertSettings merged = new InsertSettings();
+        if (source != null) {
+            merged.rawSettings.putAll(source.rawSettings);
+        }
+        if (override != null && override != source) {// avoid copying the literally same object
+            merged.rawSettings.putAll(override.rawSettings);
+        }
+        return merged;
+    }
+
+    public void setNetworkTimeout(Long networkTimeout) {
+        if (networkTimeout != null) {
+            rawSettings.put(ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.getKey(), networkTimeout.intValue());
+        } else {
+            rawSettings.remove(ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.getKey());
+        }
+    }
+
+    public Long getNetworkTimeout() {
+        return (Long) rawSettings.get(ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.getKey());
+    }
 }

client-v2/src/main/java/com/clickhouse/client/api/query/QuerySettings.java

@@ -267,6 +267,18 @@ public String getLogComment() {
         return logComment;
     }
 
+    public void setNetworkTimeout(Long networkTimeout) {
+        if (networkTimeout != null) {
+            rawSettings.put(ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.getKey(), networkTimeout.intValue());
+        } else {
+            rawSettings.remove(ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.getKey());
+        }
+    }
+
+    public Long getNetworkTimeout() {
+        return (Long) rawSettings.get(ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.getKey());
+    }
+
     public static QuerySettings merge(QuerySettings source, QuerySettings override) {
         QuerySettings merged = new QuerySettings();
         if (source != null) {

jdbc-v2/src/main/java/com/clickhouse/jdbc/ConnectionImpl.java

@@ -2,11 +2,15 @@
 
 import com.clickhouse.client.api.Client;
 import com.clickhouse.client.api.ClientConfigProperties;
+import com.clickhouse.client.api.ClientException;
+import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader;
 import com.clickhouse.client.api.internal.ServerSettings;
 import com.clickhouse.client.api.metadata.TableSchema;
 import com.clickhouse.client.api.query.GenericRecord;
+import com.clickhouse.client.api.query.QueryResponse;
 import com.clickhouse.client.api.query.QuerySettings;
 import com.clickhouse.data.ClickHouseDataType;
+import com.clickhouse.data.ClickHouseFormat;
 import com.clickhouse.jdbc.internal.ExceptionUtils;
 import com.clickhouse.jdbc.internal.JdbcConfiguration;
 import com.clickhouse.jdbc.internal.JdbcUtils;
@@ -16,6 +20,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.Closeable;
 import java.sql.Array;
 import java.sql.Blob;
 import java.sql.CallableStatement;
@@ -43,10 +48,11 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 public class ConnectionImpl implements Connection, JdbcV2Wrapper {
-    private static final Logger log = LoggerFactory.getLogger(ConnectionImpl.class);
+    private static final Logger LOG = LoggerFactory.getLogger(ConnectionImpl.class);
 
     protected final String url;
     private final Client client; // this member is private to force using getClient()
@@ -65,9 +71,11 @@ public class ConnectionImpl implements Connection, JdbcV2Wrapper {
 
     private final SqlParser sqlParser;
 
+    private Executor networkTimeoutExecutor;
+
     public ConnectionImpl(String url, Properties info) throws SQLException {
         try {
-            log.debug("Creating connection to {}", url);
+            LOG.debug("Creating connection to {}", url);
             this.url = url;//Raw URL
             this.config = new JdbcConfiguration(url, info);
             this.onCluster = false;
@@ -86,10 +94,10 @@ public ConnectionImpl(String url, Properties info) throws SQLException {
             }
 
             if (this.config.isDisableFrameworkDetection()) {
-                log.debug("Framework detection is disabled.");
+                LOG.debug("Framework detection is disabled.");
             } else {
                 String detectedFrameworks = Driver.FrameworksDetection.getFrameworksDetected();
-                log.debug("Detected frameworks: {}", detectedFrameworks);
+                LOG.debug("Detected frameworks: {}", detectedFrameworks);
                 if (!detectedFrameworks.trim().isEmpty()) {
                     clientName += " (" + detectedFrameworks + ")";
                 }
@@ -210,9 +218,8 @@ public void close() throws SQLException {
         if (isClosed()) {
             return;
         }
-
-        client.close();
-        closed = true;
+        closed = true; // mark as closed to prevent further invocations
+        client.close(); // this will disrupt pending requests.
     }
 
     @Override
@@ -597,27 +604,59 @@ public String getSchema() throws SQLException {
 
     @Override
     public void abort(Executor executor) throws SQLException {
-        if (!config.isIgnoreUnsupportedRequests()) {
-            throw new SQLFeatureNotSupportedException("abort not supported", ExceptionUtils.SQL_STATE_FEATURE_NOT_SUPPORTED);
+        if (executor == null) {
+            throw new SQLException("Executor must be not null");
         }
+        // This method should check permissions with SecurityManager but the one is deprecated.
+        // There is no replacement for SecurityManger and it is marked for removal.
+        this.close();
     }
 
     @Override
     public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
-        //TODO: Should this be supported?
-        if (!config.isIgnoreUnsupportedRequests()) {
-            throw new SQLFeatureNotSupportedException("setNetworkTimeout not supported", ExceptionUtils.SQL_STATE_FEATURE_NOT_SUPPORTED);
+        ensureOpen();
+
+        // Very good mail thread about this method implementation. https://mail.openjdk.org/pipermail/jdbc-spec-discuss/2017-November/000236.html
+
+        // This method should check permissions with SecurityManager but the one is deprecated.
+        // There is no replacement for SecurityManger and it is marked for removal.
+        if (milliseconds > 0 && executor == null) {
+            // we need executor only for positive timeout values.
+            throw new SQLException("Executor must be not null");
+        }
+        if (milliseconds < 0) {
+            throw new SQLException("Timeout must be >= 0");
+        }
+
+        // How it should work:
+        // if timeout is set with this method then any timeout exception should be reported to the connection
+        // when connection get signal about timeout it uses executor to abort itself
+        // Some connection pools set timeout before calling Connection#close() to ensure that this operation will not hang
+        // Socket timeout is propagated with QuerySettings this connection has.
+        networkTimeoutExecutor = executor;
+        defaultQuerySettings.setOption(ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.getKey(), (long)milliseconds);
+    }
+
+
+    // Should be called by child object to notify about timeout.
+    public void onNetworkTimeout() throws SQLException {
+        if (isClosed() || networkTimeoutExecutor == null) {
+            return; // we closed already so do nothing.
         }
+
+        networkTimeoutExecutor.execute(() -> {
+            try {
+                this.abort(networkTimeoutExecutor);
+            } catch (SQLException e) {
+                throw new RuntimeException("Failed to abort connection", e);
+            }
+        });
     }
 
     @Override
     public int getNetworkTimeout() throws SQLException {
-        //TODO: Should this be supported?
-        if (!config.isIgnoreUnsupportedRequests()) {
-            throw new SQLFeatureNotSupportedException("getNetworkTimeout not supported", ExceptionUtils.SQL_STATE_FEATURE_NOT_SUPPORTED);
-        }
-
-        return -1;
+        Long networkTimeout = defaultQuerySettings.getNetworkTimeout();
+        return networkTimeout == null ? 0 : networkTimeout.intValue();
     }
 
     /**

jdbc-v2/src/main/java/com/clickhouse/jdbc/ResultSetImpl.java

@@ -15,6 +15,7 @@
 import java.io.Reader;
 import java.io.StringReader;
 import java.math.BigDecimal;
+import java.net.SocketTimeoutException;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.sql.Blob;
@@ -99,6 +100,9 @@ public boolean next() throws SQLException {
         try {
             return reader.next() != null;
         } catch (Exception e) {
+            if (e instanceof SocketTimeoutException) {
+                this.parentStatement.onNetworkTimeout();
+            }
             throw ExceptionUtils.toSqlState(e);
         }
     }

jdbc-v2/src/main/java/com/clickhouse/jdbc/StatementImpl.java

@@ -11,6 +11,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.net.SocketTimeoutException;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.SQLWarning;
@@ -168,6 +169,10 @@ protected ResultSetImpl executeQueryImpl(String sql, QuerySettings settings) thr
                 }
             }
             onResultSetClosed(null);
+
+            if (e instanceof SocketTimeoutException) {
+                this.connection.onNetworkTimeout();
+            }
             throw ExceptionUtils.toSqlState(e);
         }
     }
@@ -203,6 +208,9 @@ protected long executeUpdateImpl(String sql, QuerySettings settings) throws SQLE
             updateCount = Math.max(0, (int) response.getWrittenRows()); // when statement alters schema no result rows returned.
             lastQueryId = response.getQueryId();
         } catch (Exception e) {
+            if (e instanceof SocketTimeoutException) {
+                this.connection.onNetworkTimeout();
+            }
             throw ExceptionUtils.toSqlState(e);
         }
 
@@ -610,4 +618,9 @@ public long executeLargeUpdate(String sql, String[] columnNames) throws SQLExcep
     public String getLastQueryId() {
         return lastQueryId;
     }
+
+    // Proxy method for child objects. Do not call.
+    public void onNetworkTimeout() throws SQLException {
+        this.connection.onNetworkTimeout();
+    }
 }

jdbc-v2/src/main/java/com/clickhouse/jdbc/WriterStatementImpl.java

@@ -16,6 +16,7 @@
 import java.io.InputStream;
 import java.io.Reader;
 import java.math.BigDecimal;
+import java.net.SocketTimeoutException;
 import java.net.URL;
 import java.sql.Array;
 import java.sql.Blob;
@@ -109,6 +110,9 @@ public long executeLargeUpdate() throws SQLException {
         try {
             writer.commitRow();
         } catch (Exception e) {
+            if (e instanceof SocketTimeoutException) {
+                this.connection.onNetworkTimeout();
+            }
             throw new SQLException(e);
         }
 
@@ -121,6 +125,9 @@ public long executeLargeUpdate() throws SQLException {
             updateCount = Math.max(0, (int) response.getWrittenRows()); // when statement alters schema no result rows returned.
             lastQueryId = response.getQueryId();
         } catch (Exception e) {
+            if (e instanceof SocketTimeoutException) {
+                this.connection.onNetworkTimeout();
+            }
             throw ExceptionUtils.toSqlState(e);
         } finally {
             try {
@@ -298,6 +305,9 @@ public void addBatch() throws SQLException {
         try {
             writer.commitRow();
         } catch (Exception e) {
+            if (e instanceof SocketTimeoutException) {
+                this.connection.onNetworkTimeout();
+            }
             throw new SQLException(e);
         }
     }

jdbc-v2/src/main/java/com/clickhouse/jdbc/internal/JdbcUtils.java

@@ -6,6 +6,7 @@
 import com.clickhouse.data.Tuple;
 import com.clickhouse.jdbc.types.Array;
 import com.google.common.collect.ImmutableMap;
+import org.slf4j.Logger;
 
 import java.awt.*;
 import java.math.BigInteger;
@@ -297,4 +298,14 @@ public static List<Object> convertList(List<Object> values, Class<?> type) throw
         }
         return convertedValues;
     }
+
+    public static void safeClose(AutoCloseable closeable, Logger logger) {
+        if (closeable != null) {
+            try {
+                closeable.close();
+            } catch (Exception ex) {
+                logger.warn("Failed to close closeable after exception", ex);
+            }
+        }
+    }
 }

jdbc-v2/src/test/java/com/clickhouse/jdbc/ConnectionTest.java

@@ -28,10 +28,13 @@
 import java.util.Base64;
 import java.util.Properties;
 import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 
 public class ConnectionTest extends JdbcIntegrationTest {
@@ -384,20 +387,23 @@ public void setSchemaTest() throws SQLException {
 
     @Test(groups = { "integration" })
     public void abortTest() throws SQLException {
-        Connection localConnection = this.getJdbcConnection();
-        assertThrows(SQLFeatureNotSupportedException.class, () -> localConnection.abort(null));
+        try (Connection conn = this.getJdbcConnection()) {
+            conn.abort(Executors.newSingleThreadExecutor());
+            assertTrue(conn.isClosed());
+        }
     }
 
     @Test(groups = { "integration" })
-    public void setNetworkTimeoutTest() throws SQLException {
-        Connection localConnection = this.getJdbcConnection();
-        assertThrows(SQLFeatureNotSupportedException.class, () -> localConnection.setNetworkTimeout(null, 0));
-    }
+    public void testNetworkTimeout() throws SQLException {
+        try {
+            Connection conn = this.getJdbcConnection();
+            int t1 = (int) TimeUnit.SECONDS.toMillis(20);
+            conn.setNetworkTimeout(Executors.newSingleThreadExecutor(), t1);
+            Assert.assertEquals(t1, conn.getNetworkTimeout());
 
-    @Test(groups = { "integration" })
-    public void getNetworkTimeoutTest() throws SQLException {
-        Connection localConnection = this.getJdbcConnection();
-        assertThrows(SQLFeatureNotSupportedException.class, localConnection::getNetworkTimeout);
+        } catch (Exception e) {
+
+        }
     }
 
     @Test(groups = { "integration" })