diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java index 70db8d438..1d47404b0 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java +++ b/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 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 diff --git a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java index b8b412742..85d0d737d 100644 --- a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java +++ b/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()) { diff --git a/history/latest/3203.md b/history/latest/3203.md new file mode 100644 index 000000000..052da1f6e --- /dev/null +++ b/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) diff --git a/jdbc-v2/src/test/java/com/clickhouse/jdbc/ResultSetImplTest.java b/jdbc-v2/src/test/java/com/clickhouse/jdbc/ResultSetImplTest.java index f6ad7842f..22999cf3e 100644 --- a/jdbc-v2/src/test/java/com/clickhouse/jdbc/ResultSetImplTest.java +++ b/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()) {