[client-v2, jdbc-v2] Do not fail closing a partly read response that the server ended early

client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java

@@ -43,6 +43,7 @@
 import org.apache.hc.client5.http.ssl.SSLConnectionSocketFactory;
 import org.apache.hc.client5.http.ssl.TlsSocketStrategy;
 import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.ConnectionClosedException;
 import org.apache.hc.core5.http.ConnectionRequestTimeoutException;
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.Header;
@@ -81,6 +82,7 @@
 import javax.net.ssl.SSLException;
 import javax.net.ssl.SSLParameters;
 import javax.net.ssl.SSLSocket;
+import java.io.FilterInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -658,6 +660,8 @@ private static final class TransportResponseImpl implements TransportResponse {
 
         private final ClassicHttpResponse delegate;
 
+        private boolean dataStreamCreated;
+
         TransportResponseImpl(ClassicHttpResponse delegate) {
             this.delegate = delegate;
         }
@@ -693,19 +697,67 @@ public Map<String, String> getHeaders() {
 
         @Override
         public void close() throws IOException {
-            delegate.close();
+            try {
+                delegate.close();
+            } catch (ConnectionClosedException e) {
+                // Without a data stream nobody reads the body, and this is the only report of a failed query.
+                if (!dataStreamCreated) {
+                    throw e;
+                }
+                logIncompleteBodyOnClose(getQueryId(), e);
+            }
         }
 
         @Override
         public InputStream createDataInputStream() {
             try {
-                return delegate.getEntity().getContent();
+                InputStream stream = new ResponseBodyInputStream(delegate.getEntity().getContent(), getQueryId());
+                dataStreamCreated = true;
+                return stream;
             } catch (Exception e) {
                 throw new ClientException("Failed to construct input stream", e);
             }
         }
     }
 
+    /**
+     * Closing a response body that is not read to the end makes the HTTP client read and discard the rest of it.
+     * When the server ends the body early (it does so when a query fails after the response started), that read
+     * fails with {@link ConnectionClosedException}. The caller stopped reading, so it is logged only; reads still throw.
+     */
+    private static final class ResponseBodyInputStream extends FilterInputStream {
+
+        private final String queryId;
+
+        ResponseBodyInputStream(InputStream in, String queryId) {
+            super(in);
+            this.queryId = queryId;
+        }
+
+        @Override
+        public int read() throws IOException {
+            return in.read();
+        }
+
+        @Override
+        public int read(byte[] b, int off, int len) throws IOException {
+            return in.read(b, off, len);
+        }
+
+        @Override
+        public void close() throws IOException {
+            try {
+                super.close();
+            } catch (ConnectionClosedException e) {
+                logIncompleteBodyOnClose(queryId, e);
+            }
+        }
+    }
+
+    private static void logIncompleteBodyOnClose(String queryId, ConnectionClosedException e) {
+        LOG.debug("Ignoring incomplete response body on close (queryId = {})", queryId, e);
+    }
+
     /**
      * Executes a single transport request and records it as a child span of the given operation
      * span, so a retried operation reports one request span per attempt. The request itself is

client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java

@@ -286,6 +286,65 @@ public void testClosingLZ4InputStreamReleasesConnection() throws Exception {
         }
     }
 
+    @Test(groups = {"integration"}, dataProvider = "testClosingNotFullyReadResponseDataProvider")
+    public void testClosingNotFullyReadResponse(boolean serverCompression, boolean useHttpCompression) throws Exception {
+        if (isCloud()) {
+            throw new SkipException("Test depends on how the server ends a failed response, a proxy can change it");
+        }
+        ClickHouseNode server = getServer(ClickHouseProtocol.HTTP);
+        // The query fails after the first rows are sent, so the server ends the response body early.
+        final String query = "SELECT number, randomPrintableASCII(100), throwIf(number = 100000)"
+                + " FROM numbers(200000) SETTINGS max_block_size = 1000";
+
+        try (Client client = new Client.Builder()
+                .addEndpoint(Protocol.HTTP, server.getHost(), server.getPort(), false)
+                .setUsername("default")
+                .setPassword(ClickHouseServerForTest.getPassword())
+                .setDefaultDatabase(ClickHouseServerForTest.getDatabase())
+                .compressServerResponse(serverCompression)
+                .useHttpCompression(useHttpCompression)
+                .setMaxConnections(1)
+                .setConnectionRequestTimeout(1, ChronoUnit.SECONDS)
+                .build()) {
+            QueryResponse response = client.query(query).get(10, TimeUnit.SECONDS);
+            ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(response);
+            Assert.assertNotNull(reader.next());
+            reader.close();
+            response.close();
+
+            response = client.query(query).get(10, TimeUnit.SECONDS);
+            Assert.assertNotNull(client.newBinaryFormatReader(response).next());
+            response.close();
+
+            try (QueryResponse readResponse = client.query(query).get(10, TimeUnit.SECONDS);
+                 ClickHouseBinaryFormatReader readReader = client.newBinaryFormatReader(readResponse)) {
+                Assert.expectThrows(ClientException.class, () -> {
+                    while (readReader.next() != null) {
+                        // read to the end
+                    }
+                });
+            }
+
+            // A command does not read the response body, so closing it still reports the failure.
+            Throwable thrown = Assert.expectThrows(Exception.class, () -> client.execute(query).get(10, TimeUnit.SECONDS));
+            Assert.assertNotNull(findCause(thrown, ConnectionClosedException.class), "Unexpected command failure: " + thrown);
+
+            try (QueryResponse nextResponse = client.query("SELECT 1").get(10, TimeUnit.SECONDS)) {
+                Assert.assertNotEquals(nextResponse.getInputStream().read(), -1);
+            }
+        }
+    }
+
+    @DataProvider(name = "testClosingNotFullyReadResponseDataProvider")
+    public static Object[][] testClosingNotFullyReadResponseDataProvider() {
+        return new Object[][] {
+                // serverCompression, useHttpCompression
+                {false, false},
+                {true, false},
+                {true, true},
+        };
+    }
+
     @Test(groups = {"integration"})
     public void testConnectionReuseStrategy() {
         if (isCloud()) {

history/latest/3203.md

@@ -0,0 +1,8 @@
+- **[client-v2, jdbc-v2]** Fixed `ResultSet#close()`, `QueryResponse#close()`, and closing a reader or the response
+  stream when the result was not read to the end, which could fail with `Premature end of chunk coded message body:
+  closing chunk expected`. To close a partly read response, the HTTP client reads and discards the rest of the body.
+  When a query fails after the response started, the server ends the body early, and that read failed. When the result
+  stream was opened, this failure on close is now logged at `DEBUG` and not thrown. Errors while the response is read
+  are not changed, and a command or update that does not read the result (`Client#execute`, `Statement#executeUpdate`)
+  still reports the failure.
+  (https://github.com/ClickHouse/clickhouse-java/issues/3203)

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

@@ -4,6 +4,8 @@
 import com.clickhouse.client.api.data_formats.JacksonJsonParserFactory;
 import com.clickhouse.data.ClickHouseVersion;
 import org.testng.Assert;
+import org.testng.SkipException;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 import java.io.InputStream;
@@ -316,6 +318,47 @@ public void testJsonEachRowGetObjectReturnsParserNativeArray() throws SQLExcepti
     }
 
 
+    @Test(groups = {"integration"}, dataProvider = "testCloseNotFullyReadResultSetDataProvider")
+    public void testCloseNotFullyReadResultSet(String format) throws SQLException {
+        if (isCloud()) {
+            throw new SkipException("Test depends on how the server ends a failed response, a proxy can change it");
+        }
+        Properties properties = new Properties();
+        properties.setProperty(DriverProperties.JSON_PARSER_FACTORY.getKey(), JacksonJsonParserFactory.class.getName());
+        properties.setProperty(ClientConfigProperties.INPUT_OUTPUT_FORMAT.getKey(), format);
+        properties.setProperty(ClientConfigProperties.HTTP_MAX_OPEN_CONNECTIONS.getKey(), "1");
+        // The query fails after the first rows are sent, so the server ends the response body early.
+        // Without http_write_exception_in_output_format = 0, older servers send the error as the last JSONEachRow row.
+        final String query = "SELECT number, randomPrintableASCII(100), throwIf(number = 100000)"
+                + " FROM numbers(200000) SETTINGS max_block_size = 1000, http_write_exception_in_output_format = 0";
+        try (Connection conn = getJdbcConnection(properties); Statement stmt = conn.createStatement()) {
+            ResultSet rs = stmt.executeQuery(query);
+            assertTrue(rs.next());
+            rs.close();
+
+            try (ResultSet readRs = stmt.executeQuery(query)) {
+                Assert.expectThrows(SQLException.class, () -> {
+                    while (readRs.next()) {
+                        // read to the end
+                    }
+                });
+            }
+
+            // An update does not read the response body, so closing it still reports the failure.
+            Assert.expectThrows(SQLException.class, () -> stmt.executeUpdate(query));
+
+            try (ResultSet next = stmt.executeQuery("SELECT 1")) {
+                assertTrue(next.next());
+                assertEquals(next.getInt(1), 1);
+            }
+        }
+    }
+
+    @DataProvider(name = "testCloseNotFullyReadResultSetDataProvider")
+    public static Object[][] testCloseNotFullyReadResultSetDataProvider() {
+        return new Object[][] {{"RowBinaryWithNamesAndTypes"}, {"JSONEachRow"}};
+    }
+
     @Test(groups = {"integration"})
     public void testFetchDirectionsAndSize() throws SQLException {
         try (Connection conn = getJdbcConnection(); Statement stmt = conn.createStatement()) {