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

CHANGELOG.md

@@ -258,6 +258,16 @@
   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`. Normal reads remain demand-driven so parsing
+  response metadata does not drain a small response and return its HTTP connection to the pool prematurely.
+  Complete exception frames are validated by message byte length and tag even when HTTP framing is interrupted;
+  binary readers deliver the last complete row before reporting a prefetch failure, and retain server error codes.
+  (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/data_formats/internal/AbstractBinaryFormatReader.java

@@ -71,7 +71,8 @@ public abstract class AbstractBinaryFormatReader implements ClickHouseBinaryForm
     private TableSchema schema;
     private ClickHouseColumn[] columns;
     private Map[] convertions;
-    private boolean hasNext = true;
+    private boolean hasNext = true;
+    private RuntimeException nextReadException;
     private boolean initialState = true; // reader is in initial state, no records have been read yet
     private long row = -1; // before first row
     private long lastNextCallTs; // for exception to detect slow reader
@@ -226,7 +227,10 @@ public <T> T readValue(String colName) {
     }
 
     @Override
-    public boolean hasNext() {
+    public boolean hasNext() {
+        if (nextReadException != null) {
+            throw nextReadException;
+        }
         if (initialState) {
             readNextRecord();
         }
@@ -264,7 +268,10 @@ private String recordReadExceptionMsg(String column) {
     }
 
     @Override
-    public Map<String, Object> next() {
+    public Map<String, Object> next() {
+        if (nextReadException != null) {
+            throw nextReadException;
+        }
         if (!hasNext) {
             return null;
         }
@@ -274,12 +281,12 @@ public Map<String, Object> next() {
                 Object[] tmp = currentRecord;
                 currentRecord = nextRecord;
                 nextRecord = tmp;
-                readNextRecord();
+                prefetchNextRecord();
                 return new RecordWrapper(currentRecord, schema);
             } else {
                 try {
                     if (readRecord(currentRecord)) {
-                        readNextRecord();
+                        prefetchNextRecord();
                         return new RecordWrapper(currentRecord, schema);
                     } else {
                         currentRecord = null;
@@ -295,7 +302,16 @@ public Map<String, Object> next() {
         }
     }
 
-    protected void endReached() {
+    private void prefetchNextRecord() {
+        try {
+            readNextRecord();
+        } catch (RuntimeException e) {
+            // The current row is complete; report a lookahead failure only when the caller advances again.
+            nextReadException = e;
+        }
+    }
+
+    protected void endReached() {
         initialState = false;
         hasNext = false;
     }

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

@@ -1,6 +1,7 @@
 package com.clickhouse.client.api.data_formats.internal;
 
-import com.clickhouse.client.api.ClientException;
+import com.clickhouse.client.api.ClientException;
+import com.clickhouse.client.api.ServerException;
 import com.clickhouse.client.api.DataTypeUtils;
 import com.clickhouse.client.api.query.NullValueException;
 import com.clickhouse.data.ClickHouseColumn;
@@ -280,7 +281,7 @@ private <T> T readValue(ClickHouseColumn column, Class<?> typeHint, boolean stri
                 default:
                     throw new IllegalArgumentException("Unsupported data type: " + actualColumn.getDataType());
             }
-        } catch (EOFException e) {
+        } catch (EOFException | ServerException e) {
             throw e;
         } catch (Exception e) {
             log.debug("Failed to read value for column {}, {}", column.getColumnName(), e.getLocalizedMessage());

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/ClickHouseLZ4InputStream.java

@@ -54,14 +54,15 @@ public int read(byte[] b, int off, int len) throws IOException {
             return 0;
         }
 
-        int readBytes = 0;
-        do {
-            int remaining = Math.min(len - readBytes, buffer.remaining());
-            buffer.get(b, off + readBytes, remaining);
-            readBytes += remaining;
-        } while (readBytes < len && refill() != -1);
-
-        return readBytes == 0 ? -1 : readBytes;
+        while (!buffer.hasRemaining()) {
+            if (refill() == -1) {
+                return -1;
+            }
+        }
+        // Return decoded bytes before trying another block, which may end in a transport error.
+        int readBytes = Math.min(len, buffer.remaining());
+        buffer.get(b, off, readBytes);
+        return readBytes;
     }
 
 

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

@@ -63,7 +63,8 @@
 import org.apache.hc.core5.http.io.entity.ByteArrayEntity;
 import org.apache.hc.core5.http.io.entity.EntityTemplate;
 import org.apache.hc.core5.http.protocol.HttpContext;
-import org.apache.hc.core5.io.CloseMode;
+import org.apache.hc.core5.io.CloseMode;
+import org.apache.hc.core5.io.ModalCloseable;
 import org.apache.hc.core5.io.IOCallback;
 import org.apache.hc.core5.net.URIAuthority;
 import org.apache.hc.core5.net.URIBuilder;
@@ -653,9 +654,10 @@ public TransportRequest createRequest(Endpoint server, Map<String, Object> reque
     }
 
 
-    private static final class TransportResponseImpl implements TransportResponse {
-
-        private final ClassicHttpResponse delegate;
+    private static final class TransportResponseImpl implements TransportResponse {
+
+        private final ClassicHttpResponse delegate;
+        private volatile boolean aborted;
 
         TransportResponseImpl(ClassicHttpResponse delegate) {
             this.delegate = delegate;
@@ -691,14 +693,27 @@ public Map<String, String> getHeaders() {
         }
 
         @Override
-        public void close() throws IOException {
-            delegate.close();
+        public void close() throws IOException {
+            if (!aborted) {
+                delegate.close();
+            }
         }
 
         @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(),
+                                () -> {
+                                    // A tagged exception can deliberately leave HTTP chunk framing incomplete.
+                                    if (delegate instanceof ModalCloseable) {
+                                        aborted = true;
+                                        ((ModalCloseable) delegate).close(CloseMode.IMMEDIATE);
+                                    }
+                                });
             } catch (Exception e) {
                 throw new ClientException("Failed to construct input stream", e);
             }
@@ -1066,7 +1081,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
     ));
 
     /**