fix(client-v2,jdbc-v2): handle tagged errors in response streams

CHANGELOG.md

@@ -142,6 +142,13 @@
   engine-to-table-type mapping, so it fell back to the default `TABLE`, and `getTables(..., types = {"REMOTE TABLE"})`
   returned no row for such a table. `BigQuery` is now mapped to `REMOTE TABLE`, like the other external-storage
   engines. (https://github.com/ClickHouse/clickhouse-java/issues/3049)
+- **[client-v2, jdbc-v2]** Fixed ClickHouse exceptions appended to an HTTP 200 response body being exposed as result
+  data while a query was streamed. `client-v2` now authenticates the in-band exception frame with the
+  `X-ClickHouse-Exception-Tag` response header and throws a `ServerException` when the stream reaches it. When the
+  server error is `TIMEOUT_EXCEEDED` (code 159), `jdbc-v2` now reports `SQLTimeoutException` with SQLState `HYT00`
+  from `ResultSet.next()` while preserving the original exception chain. Previously the tagged frame could be read as
+  row data and the timeout was exposed as a generic `SQLException`.
+  (https://github.com/ClickHouse/clickhouse-java/issues/2702, https://github.com/ClickHouse/clickhouse-java/issues/3077)
 - **[jdbc-v2]** Fixed `PreparedStatement.getMetaData()` losing the result-set schema for a statement whose SQL
   contains a comment. The `DESCRIBE` query used to resolve the metadata was built by re-scanning the SQL with a
   regex that knew only quoted tokens, so a `?` inside a `--` / `#` / `/* */` comment was rewritten to `NULL` and

client-v2/src/main/java/com/clickhouse/client/api/http/ClickHouseHttpProto.java

@@ -27,6 +27,12 @@ public class ClickHouseHttpProto {
      */
     public static final String HEADER_EXCEPTION_CODE = "X-ClickHouse-Exception-Code";
 
+    /**
+     * Response only header containing the tag used to identify exception frames in a successful response body.
+     * Cannot be used in request.
+     */
+    public static final String HEADER_EXCEPTION_TAG = "X-ClickHouse-Exception-Tag";
+
     /**
      * Response only header to indicate a query progress.
      * Cannot be used in request.

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

@@ -693,7 +693,11 @@ public void close() throws IOException {
         @Override
         public InputStream createDataInputStream() {
             try {
-                return delegate.getEntity().getContent();
+                InputStream input = delegate.getEntity().getContent();
+                Header exceptionTag = delegate.getFirstHeader(ClickHouseHttpProto.HEADER_EXCEPTION_TAG);
+                return exceptionTag == null || exceptionTag.getValue().isEmpty()
+                        ? input
+                        : new HttpExceptionInputStream(input, exceptionTag.getValue(), delegate.getCode(), getQueryId());
             } catch (Exception e) {
                 throw new ClientException("Failed to construct input stream", e);
             }
@@ -1055,7 +1059,8 @@ public static int getHeaderInt(Header header, int defaultValue) {
             ClickHouseHttpProto.HEADER_DB_USER,
             ClickHouseHttpProto.HEADER_TIMEZONE,
             ClickHouseHttpProto.HEADER_FORMAT,
-            ClickHouseHttpProto.HEADER_PROGRESS
+            ClickHouseHttpProto.HEADER_PROGRESS,
+            ClickHouseHttpProto.HEADER_EXCEPTION_TAG
     ));
 
     /**

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

@@ -0,0 +1,253 @@
+package com.clickhouse.client.api.internal;
+
+import com.clickhouse.client.api.ClientException;
+import com.clickhouse.client.api.ServerException;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Hides tagged exception frames appended to successful HTTP response bodies. A possible frame prefix remains buffered
+ * until it is matched or disproved, so callers never observe part of a marker when it crosses source read boundaries.
+ */
+final class HttpExceptionInputStream extends InputStream {
+
+    private static final byte[] EXCEPTION_MARKER = "\r\n__exception__\r\n".getBytes(StandardCharsets.UTF_8);
+    private static final String EXCEPTION_END_MARKER = "\r\n__exception__\r\n";
+    private static final int BUFFER_SIZE = 8192;
+    private static final int MAX_EXCEPTION_SIZE = 32 * 1024;
+    private static final Pattern ERROR_CODE_PATTERN = Pattern.compile("^Code:\\s*(\\d+)\\.");
+
+    private final InputStream source;
+    private final String exceptionTag;
+    private final int transportStatus;
+    private final String queryId;
+    private final byte[] exceptionPrefix;
+    private final byte[] sourceBuffer = new byte[BUFFER_SIZE];
+
+    private byte[] pending = new byte[BUFFER_SIZE];
+    private int pendingStart;
+    private int pendingEnd;
+    private int scanOffset;
+    private boolean sourceDone;
+    private RuntimeException terminalException;
+    private IOException terminalIOException;
+
+    HttpExceptionInputStream(InputStream source, String exceptionTag, int transportStatus, String queryId) {
+        this.source = source;
+        this.exceptionTag = exceptionTag;
+        this.transportStatus = transportStatus;
+        this.queryId = queryId;
+        byte[] tagBytes = exceptionTag.getBytes(StandardCharsets.UTF_8);
+        this.exceptionPrefix = Arrays.copyOf(EXCEPTION_MARKER, EXCEPTION_MARKER.length + tagBytes.length + 2);
+        System.arraycopy(tagBytes, 0, exceptionPrefix, EXCEPTION_MARKER.length, tagBytes.length);
+        exceptionPrefix[exceptionPrefix.length - 2] = '\r';
+        exceptionPrefix[exceptionPrefix.length - 1] = '\n';
+    }
+
+    @Override
+    public int read() throws IOException {
+        byte[] oneByte = new byte[1];
+        int read = read(oneByte, 0, 1);
+        return read < 0 ? -1 : oneByte[0] & 0xff;
+    }
+
+    @Override
+    public int read(byte[] buffer, int offset, int length) throws IOException {
+        if (buffer == null) {
+            throw new NullPointerException("buffer");
+        }
+        if (offset < 0 || length < 0 || length > buffer.length - offset) {
+            throw new IndexOutOfBoundsException();
+        }
+        if (length == 0) {
+            return 0;
+        }
+
+        while (true) {
+            int safeLength = safeLength();
+            if (safeLength > 0) {
+                int read = Math.min(length, safeLength);
+                System.arraycopy(pending, pendingStart, buffer, offset, read);
+                pendingStart += read;
+                return read;
+            }
+            if (terminalException != null) {
+                throw terminalException;
+            }
+            if (terminalIOException != null) {
+                throw terminalIOException;
+            }
+            if (sourceDone) {
+                return -1;
+            }
+
+            fillPending();
+            scanPending();
+        }
+    }
+
+    @Override
+    public int available() {
+        return safeLength();
+    }
+
+    @Override
+    public void close() throws IOException {
+        source.close();
+    }
+
+    private int safeLength() {
+        if (sourceDone || terminalException != null || terminalIOException != null) {
+            return pendingEnd - pendingStart;
+        }
+        return Math.max(0, scanOffset - pendingStart);
+    }
+
+    private void fillPending() {
+        try {
+            int read = source.read(sourceBuffer);
+            if (read < 0) {
+                sourceDone = true;
+                scanOffset = pendingEnd;
+                return;
+            }
+            appendPending(sourceBuffer, read);
+        } catch (IOException e) {
+            sourceDone = true;
+            terminalIOException = e;
+            scanOffset = pendingEnd;
+        }
+    }
+
+    private void appendPending(byte[] bytes, int length) {
+        compactPending(length);
+        System.arraycopy(bytes, 0, pending, pendingEnd, length);
+        pendingEnd += length;
+    }
+
+    private void compactPending(int additionalLength) {
+        int currentLength = pendingEnd - pendingStart;
+        if (pending.length - pendingEnd >= additionalLength) {
+            return;
+        }
+
+        int newLength = Math.max(pending.length * 2, currentLength + additionalLength);
+        byte[] compacted = new byte[newLength];
+        System.arraycopy(pending, pendingStart, compacted, 0, currentLength);
+        scanOffset -= pendingStart;
+        pendingStart = 0;
+        pendingEnd = currentLength;
+        pending = compacted;
+    }
+
+    private void scanPending() {
+        int exceptionStart = indexOf(pending, scanOffset, pendingEnd, exceptionPrefix);
+        if (exceptionStart >= 0) {
+            captureException(exceptionStart);
+            return;
+        }
+
+        int suffixLength = matchingSuffixLength(pending, pendingStart, pendingEnd, exceptionPrefix);
+        scanOffset = pendingEnd - suffixLength;
+    }
+
+    private void captureException(int exceptionStart) {
+        ByteArrayOutputStream exceptionBody = new ByteArrayOutputStream();
+        int bodyStart = exceptionStart + exceptionPrefix.length;
+        exceptionBody.write(pending, bodyStart, pendingEnd - bodyStart);
+        pendingEnd = exceptionStart;
+        scanOffset = exceptionStart;
+
+        try {
+            while (exceptionBody.size() <= MAX_EXCEPTION_SIZE) {
+                int read = source.read(sourceBuffer);
+                if (read < 0) {
+                    terminalException = parseException(exceptionBody.toByteArray());
+                    sourceDone = true;
+                    return;
+                }
+                int remaining = MAX_EXCEPTION_SIZE + 1 - exceptionBody.size();
+                exceptionBody.write(sourceBuffer, 0, Math.min(read, remaining));
+                if (read > remaining || exceptionBody.size() > MAX_EXCEPTION_SIZE) {
+                    terminalException = new ClientException("ClickHouse exception frame exceeds "
+                            + MAX_EXCEPTION_SIZE + " bytes");
+                    sourceDone = true;
+                    return;
+                }
+            }
+        } catch (IOException e) {
+            ClientException truncatedFrame = new ClientException(
+                    "Failed to finish reading ClickHouse exception frame", parseException(exceptionBody.toByteArray()));
+            truncatedFrame.addSuppressed(e);
+            terminalException = truncatedFrame;
+            sourceDone = true;
+        }
+    }
+
+    private ServerException parseException(byte[] body) {
+        String message = stripTrailer(new String(body, StandardCharsets.UTF_8)).trim();
+        Matcher matcher = ERROR_CODE_PATTERN.matcher(message);
+        int errorCode = matcher.find() ? Integer.parseInt(matcher.group(1)) : ServerException.CODE_UNKNOWN;
+        return new ServerException(errorCode, message, transportStatus, queryId);
+    }
+
+    private String stripTrailer(String body) {
+        int closingMarker = body.lastIndexOf(EXCEPTION_END_MARKER);
+        if (closingMarker < 0) {
+            return body;
+        }
+
+        String beforeMarker = body.substring(0, closingMarker);
+        int trailerStart = beforeMarker.lastIndexOf("\r\n");
+        if (trailerStart < 0) {
+            return body;
+        }
+
+        String trailer = beforeMarker.substring(trailerStart + 2);
+        int separator = trailer.indexOf(' ');
+        if (separator <= 0 || !trailer.substring(separator + 1).equals(exceptionTag)) {
+            return body;
+        }
+        for (int i = 0; i < separator; i++) {
+            if (!Character.isDigit(trailer.charAt(i))) {
+                return body;
+            }
+        }
+        return beforeMarker.substring(0, trailerStart);
+    }
+
+    private static int indexOf(byte[] data, int from, int to, byte[] pattern) {
+        int lastStart = to - pattern.length;
+        for (int i = from; i <= lastStart; i++) {
+            int j = 0;
+            while (j < pattern.length && data[i + j] == pattern[j]) {
+                j++;
+            }
+            if (j == pattern.length) {
+                return i;
+            }
+        }
+        return -1;
+    }
+
+    private static int matchingSuffixLength(byte[] data, int from, int to, byte[] pattern) {
+        int maxLength = Math.min(pattern.length - 1, to - from);
+        for (int length = maxLength; length > 0; length--) {
+            int suffixStart = to - length;
+            int i = 0;
+            while (i < length && data[suffixStart + i] == pattern[i]) {
+                i++;
+            }
+            if (i == length) {
+                return length;
+            }
+        }
+        return 0;
+    }
+}

client-v2/src/test/java/com/clickhouse/client/api/HttpResponseExceptionTest.java

@@ -0,0 +1,93 @@
+package com.clickhouse.client.api;
+
+import com.clickhouse.client.api.http.ClickHouseHttpProto;
+import com.clickhouse.client.api.internal.ClickHouseLZ4OutputStream;
+import com.clickhouse.client.api.query.QueryResponse;
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.client.WireMock;
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
+import net.jpountz.lz4.LZ4Factory;
+import org.apache.hc.core5.http.HttpStatus;
+import org.testng.Assert;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeUnit;
+
+public class HttpResponseExceptionTest {
+
+    @DataProvider(name = "responseCompression")
+    public Object[][] responseCompression() {
+        return new Object[][] {{false}, {true}};
+    }
+
+    @Test(dataProvider = "responseCompression")
+    public void shouldThrowServerExceptionWhileReadingSuccessfulResponse(boolean compressedResponse) throws Exception {
+        String exceptionTag = "0123456789abcdef";
+        String queryId = "mid-stream-timeout";
+        byte[] resultPrefix = "result-data".getBytes(StandardCharsets.UTF_8);
+        String errorMessage = "Code: 159. DB::Exception: Timeout exceeded. (TIMEOUT_EXCEEDED)";
+        String exceptionFrame = "\r\n__exception__\r\n" + exceptionTag + "\r\n" + errorMessage
+                + "\r\n" + errorMessage.getBytes(StandardCharsets.UTF_8).length + " " + exceptionTag
+                + "\r\n__exception__\r\n";
+        byte[] body = responseBody(resultPrefix, exceptionFrame.getBytes(StandardCharsets.UTF_8), compressedResponse);
+
+        WireMockServer mockServer = new WireMockServer(WireMockConfiguration.options().dynamicPort());
+        mockServer.start();
+
+        try {
+            mockServer.stubFor(WireMock.post(WireMock.anyUrl())
+                    .willReturn(WireMock.aResponse()
+                            .withStatus(HttpStatus.SC_OK)
+                            .withHeader("X-ClickHouse-Exception-Tag", exceptionTag)
+                            .withHeader(ClickHouseHttpProto.HEADER_QUERY_ID, queryId)
+                            .withBody(body)));
+
+            try (Client client = new Client.Builder()
+                    .addEndpoint("http://localhost:" + mockServer.port())
+                    .setUsername("default")
+                    .setPassword("")
+                    .setDefaultDatabase("default")
+                    .compressServerResponse(compressedResponse)
+                    .useHttpCompression(false)
+                    .build();
+                 QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS);
+                 InputStream input = response.getInputStream()) {
+                byte[] actualPrefix = new byte[resultPrefix.length];
+                new DataInputStream(input).readFully(actualPrefix);
+                Assert.assertEquals(actualPrefix, resultPrefix);
+
+                ServerException exception = Assert.expectThrows(ServerException.class, input::read);
+                Assert.assertEquals(exception.getCode(), 159);
+                Assert.assertEquals(exception.getTransportProtocolCode(), HttpStatus.SC_OK);
+                Assert.assertEquals(exception.getQueryId(), queryId);
+                Assert.assertTrue(exception.getMessage().startsWith(errorMessage), exception.getMessage());
+            }
+        } finally {
+            mockServer.stop();
+        }
+    }
+
+    private static byte[] responseBody(byte[] resultPrefix, byte[] exceptionFrame, boolean compressed)
+            throws IOException {
+        ByteArrayOutputStream body = new ByteArrayOutputStream();
+        if (!compressed) {
+            body.write(resultPrefix);
+            body.write(exceptionFrame);
+            return body.toByteArray();
+        }
+
+        try (ClickHouseLZ4OutputStream output = new ClickHouseLZ4OutputStream(body,
+                LZ4Factory.fastestInstance().fastCompressor(), ClickHouseLZ4OutputStream.UNCOMPRESSED_BUFF_SIZE)) {
+            output.write(resultPrefix);
+            output.flush();
+            output.write(exceptionFrame);
+        }
+        return body.toByteArray();
+    }
+}