diff --git a/CHANGELOG.md b/CHANGELOG.md
index dfba0c644..ba05b2588 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/AbstractBinaryFormatReader.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/AbstractBinaryFormatReader.java
index 6c175e2f4..dbd6259fb 100644
--- a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/AbstractBinaryFormatReader.java
+++ b/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;
     }
diff --git a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/BinaryStreamReader.java b/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/BinaryStreamReader.java
index dbdbd559b..4bf9653da 100644
--- a/client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/BinaryStreamReader.java
+++ b/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());
diff --git a/client-v2/src/main/java/com/clickhouse/client/api/http/ClickHouseHttpProto.java b/client-v2/src/main/java/com/clickhouse/client/api/http/ClickHouseHttpProto.java
index 860f74ca6..506af93f1 100644
--- a/client-v2/src/main/java/com/clickhouse/client/api/http/ClickHouseHttpProto.java
+++ b/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.
diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/ClickHouseLZ4InputStream.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/ClickHouseLZ4InputStream.java
index 980b06cf9..518e17645 100644
--- a/client-v2/src/main/java/com/clickhouse/client/api/internal/ClickHouseLZ4InputStream.java
+++ b/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;
     }
 
 
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 9dab47a39..17473de1e 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
@@ -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
     ));
 
     /**
diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpExceptionInputStream.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpExceptionInputStream.java
new file mode 100644
index 000000000..ec30e6478
--- /dev/null
+++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpExceptionInputStream.java
@@ -0,0 +1,276 @@
+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 Runnable onCompleteException;
+    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, exceptionTag, transportStatus, queryId, () -> { });
+    }
+
+    HttpExceptionInputStream(InputStream source, String exceptionTag, int transportStatus, String queryId,
+                             Runnable onCompleteException) {
+        this.source = source;
+        this.exceptionTag = exceptionTag;
+        this.transportStatus = transportStatus;
+        this.queryId = queryId;
+        this.onCompleteException = onCompleteException;
+        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(length);
+            scanPending();
+        }
+    }
+
+    @Override
+    public int available() {
+        return safeLength();
+    }
+
+    @Override
+    public void close() throws IOException {
+        try {
+            source.close();
+        } catch (IOException e) {
+            if (!(terminalException instanceof ServerException)) {
+                throw e;
+            }
+            terminalException.addSuppressed(e);
+        }
+    }
+
+    private int safeLength() {
+        if (sourceDone || terminalException != null || terminalIOException != null) {
+            return pendingEnd - pendingStart;
+        }
+        return Math.max(0, scanOffset - pendingStart);
+    }
+
+    private void fillPending(int requestedLength) {
+        try {
+            // Reading ahead can drain a small HTTP response and release its pooled connection prematurely.
+            int read = source.read(sourceBuffer, 0, Math.min(requestedLength, sourceBuffer.length));
+            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) {
+                byte[] body = exceptionBody.toByteArray();
+                int messageLength = completeMessageLength(body);
+                if (messageLength >= 0) {
+                    terminalException = parseException(Arrays.copyOf(body, messageLength));
+                    sourceDone = true;
+                    onCompleteException.run();
+                    return;
+                }
+                int read = source.read(sourceBuffer);
+                if (read < 0) {
+                    terminalException = new ClientException("Incomplete ClickHouse exception frame", parseException(body));
+                    sourceDone = true;
+                    return;
+                }
+                int remaining = MAX_EXCEPTION_SIZE + 1 - exceptionBody.size();
+                exceptionBody.write(sourceBuffer, 0, Math.min(read, remaining));
+            }
+            terminalException = new ClientException("ClickHouse exception frame exceeds " + MAX_EXCEPTION_SIZE + " bytes");
+            sourceDone = true;
+        } 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 = 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 int completeMessageLength(byte[] body) {
+        byte[] suffix = (" " + exceptionTag + EXCEPTION_END_MARKER).getBytes(StandardCharsets.UTF_8);
+        int digitsEnd = body.length - suffix.length;
+        if (digitsEnd <= 0) {
+            return -1;
+        }
+        for (int i = 0; i < suffix.length; i++) {
+            if (body[digitsEnd + i] != suffix[i]) {
+                return -1;
+            }
+        }
+        int digitsStart = digitsEnd;
+        while (digitsStart > 0 && body[digitsStart - 1] != '\n') {
+            digitsStart--;
+        }
+        if (digitsStart == 0 || digitsStart == digitsEnd) {
+            return -1;
+        }
+        int messageLength = 0;
+        for (int i = digitsStart; i < digitsEnd; i++) {
+            if (body[i] < '0' || body[i] > '9' || messageLength > MAX_EXCEPTION_SIZE / 10) {
+                return -1;
+            }
+            messageLength = messageLength * 10 + body[i] - '0';
+        }
+        // The server counts UTF-8 bytes, including the message's final newline, not Java characters.
+        return messageLength == digitsStart ? messageLength : -1;
+    }
+
+    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;
+    }
+}
diff --git a/client-v2/src/test/java/com/clickhouse/client/api/HttpResponseExceptionTest.java b/client-v2/src/test/java/com/clickhouse/client/api/HttpResponseExceptionTest.java
new file mode 100644
index 000000000..a0ecd1b2d
--- /dev/null
+++ b/client-v2/src/test/java/com/clickhouse/client/api/HttpResponseExceptionTest.java
@@ -0,0 +1,162 @@
+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.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+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)\n";
+        String exceptionFrame = "\r\n__exception__\r\n" + exceptionTag + "\r\n" + errorMessage
+                + 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.trim()), exception.getMessage());
+            }
+        } finally {
+            mockServer.stop();
+        }
+    }
+
+    @DataProvider(name = "abortedResponses")
+    public Object[][] abortedResponses() {
+        return new Object[][] {{159, false}, {60, false}, {159, true}, {60, true}};
+    }
+
+    @Test(dataProvider = "abortedResponses")
+    public void shouldRecognizeCompleteFrameWithoutFinalHttpChunk(int code, boolean compressed) throws Exception {
+        String tag = "0123456789abcdef";
+        String message = "Code: " + code + ". DB::Exception: Query failed — 错误.\n";
+        String frame = "\r\n__exception__\r\n" + tag + "\r\n" + message
+                + message.getBytes(StandardCharsets.UTF_8).length + " " + tag + "\r\n__exception__\r\n";
+        byte[] prefix = "result-data".getBytes(StandardCharsets.UTF_8);
+        byte[] body = responseBody(prefix, frame.getBytes(StandardCharsets.UTF_8), compressed);
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        try (ServerSocket server = new ServerSocket(0, 1, java.net.InetAddress.getLoopbackAddress())) {
+            server.setSoTimeout(10000);
+            Future<?> sent = executor.submit(() -> {
+                try (Socket socket = server.accept()) {
+                    socket.setSoTimeout(10000);
+                    BufferedReader request = new BufferedReader(new InputStreamReader(
+                            socket.getInputStream(), StandardCharsets.US_ASCII));
+                    int contentLength = 0;
+                    String line;
+                    while ((line = request.readLine()) != null && !line.isEmpty()) {
+                        if (line.regionMatches(true, 0, "Content-Length:", 0, 15)) {
+                            contentLength = Integer.parseInt(line.substring(15).trim());
+                        }
+                    }
+                    for (int i = 0; i < contentLength; i++) {
+                        if (request.read() < 0) {
+                            throw new IOException("Incomplete request");
+                        }
+                    }
+                    OutputStream output = socket.getOutputStream();
+                    output.write(("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n"
+                            + "X-ClickHouse-Exception-Tag: " + tag + "\r\n\r\n"
+                            + Integer.toHexString(body.length) + "\r\n").getBytes(StandardCharsets.US_ASCII));
+                    output.write(body);
+                    output.write("\r\n".getBytes(StandardCharsets.US_ASCII));
+                    output.flush(); // Deliberately close without the terminating HTTP chunk.
+                }
+                return null;
+            });
+            try (Client client = new Client.Builder().addEndpoint("http://localhost:" + server.getLocalPort())
+                    .compressServerResponse(compressed).useHttpCompression(false).build();
+                 QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS);
+                 InputStream input = response.getInputStream()) {
+                byte[] actual = new byte[prefix.length];
+                new DataInputStream(input).readFully(actual);
+                Assert.assertEquals(actual, prefix);
+                ServerException exception = Assert.expectThrows(ServerException.class, input::read);
+                Assert.assertEquals(exception.getCode(), code);
+                Assert.assertEquals(exception.getMessage(), message.trim());
+            }
+            sent.get(10, TimeUnit.SECONDS);
+        } finally {
+            executor.shutdownNow();
+            Assert.assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
+        }
+    }
+
+    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();
+    }
+}
diff --git a/client-v2/src/test/java/com/clickhouse/client/api/internal/ClickHouseLZ4InputStreamTest.java b/client-v2/src/test/java/com/clickhouse/client/api/internal/ClickHouseLZ4InputStreamTest.java
index e6ab846e5..929a0c524 100644
--- a/client-v2/src/test/java/com/clickhouse/client/api/internal/ClickHouseLZ4InputStreamTest.java
+++ b/client-v2/src/test/java/com/clickhouse/client/api/internal/ClickHouseLZ4InputStreamTest.java
@@ -1,6 +1,7 @@
 package com.clickhouse.client.api.internal;
 
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 
 import net.jpountz.lz4.LZ4Factory;
@@ -9,6 +10,24 @@
 
 public class ClickHouseLZ4InputStreamTest {
 
+    @Test
+    public void returnsDecodedBlockBeforeFailingOnNextHeader() throws Exception {
+        ByteArrayOutputStream encoded = new ByteArrayOutputStream();
+        try (ClickHouseLZ4OutputStream output = new ClickHouseLZ4OutputStream(encoded,
+                LZ4Factory.fastestJavaInstance().fastCompressor(), 8192)) {
+            output.write(new byte[] {1, 2, 3});
+        }
+        encoded.write(new byte[10]);
+        try (ClickHouseLZ4InputStream input = new ClickHouseLZ4InputStream(
+                new ByteArrayInputStream(encoded.toByteArray()),
+                LZ4Factory.fastestJavaInstance().fastDecompressor(), 8192)) {
+            byte[] buffer = new byte[64];
+            Assert.assertEquals(input.read(buffer), 3);
+            Assert.assertEquals(java.util.Arrays.copyOf(buffer, 3), new byte[] {1, 2, 3});
+            Assert.expectThrows(IOException.class, () -> input.read(buffer));
+        }
+    }
+
     @Test
     public void reportsActualByteCountsForTruncatedHeader() {
         byte[] truncatedHeader = new byte[10];
diff --git a/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpExceptionInputStreamTest.java b/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpExceptionInputStreamTest.java
new file mode 100644
index 000000000..f0683fbff
--- /dev/null
+++ b/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpExceptionInputStreamTest.java
@@ -0,0 +1,194 @@
+package com.clickhouse.client.api.internal;
+
+import com.clickhouse.client.api.ClientException;
+import com.clickhouse.client.api.ServerException;
+import com.clickhouse.client.api.ClientConfigProperties;
+import com.clickhouse.client.api.data_formats.RowBinaryWithNamesAndTypesFormatReader;
+import com.clickhouse.client.api.data_formats.internal.BinaryStreamReader;
+import com.clickhouse.client.api.query.QuerySettings;
+import org.testng.Assert;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.TimeZone;
+
+public class HttpExceptionInputStreamTest {
+
+    private static final String EXCEPTION_TAG = "0123456789abcdef";
+    private static final String ERROR_MESSAGE =
+            "Code: 159. DB::Exception: Timeout exceeded. (TIMEOUT_EXCEEDED)\n";
+
+    @DataProvider(name = "iterationModes")
+    public Object[][] iterationModes() {
+        return new Object[][] {{false}, {true}};
+    }
+
+    @Test(dataProvider = "iterationModes")
+    public void shouldDeliverLastRowBeforePrefetchFailure(boolean useHasNext) throws Exception {
+        byte[] rows = {1, 1, 'v', 5, 'U', 'I', 'n', 't', '8', 1, 2};
+        QuerySettings settings = new QuerySettings().setOption(
+                ClientConfigProperties.USE_TIMEZONE.getKey(), TimeZone.getTimeZone("UTC"));
+        try (InputStream input = new HttpExceptionInputStream(new ByteArrayInputStream(
+                responseBody(rows, exceptionFrame(EXCEPTION_TAG))), EXCEPTION_TAG, 200, "query-id");
+             RowBinaryWithNamesAndTypesFormatReader reader = new RowBinaryWithNamesAndTypesFormatReader(
+                     input, settings, new BinaryStreamReader.DefaultByteBufferAllocator())) {
+            for (int expected = 1; expected <= 2; expected++) {
+                if (useHasNext) {
+                    Assert.assertTrue(reader.hasNext());
+                }
+                Assert.assertEquals(((Number) reader.next().get("v")).intValue(), expected);
+            }
+            ServerException exception = useHasNext
+                    ? Assert.expectThrows(ServerException.class, reader::hasNext)
+                    : Assert.expectThrows(ServerException.class, reader::next);
+            Assert.assertEquals(exception.getCode(), 159);
+            Assert.assertSame(Assert.expectThrows(ServerException.class, reader::next), exception);
+        }
+    }
+
+    @DataProvider(name = "smallReads")
+    public Object[][] smallReads() {
+        return new Object[][] {{0}, {1}, {8}};
+    }
+
+    @Test(dataProvider = "smallReads")
+    public void shouldNotDrainResponseOnSmallRead(int readSize) throws Exception {
+        byte[] body = "result-data-that-must-remain-unread".getBytes(StandardCharsets.UTF_8);
+        ByteArrayInputStream source = new ByteArrayInputStream(body);
+        try (InputStream input = new HttpExceptionInputStream(source, EXCEPTION_TAG, 200, "query-id")) {
+            if (readSize == 0) {
+                Assert.assertEquals(input.read(), body[0]);
+            } else {
+                byte[] buffer = new byte[readSize];
+                Assert.assertEquals(input.read(buffer), readSize);
+                for (int i = 0; i < buffer.length; i++) {
+                    Assert.assertEquals(buffer[i], body[i]);
+                }
+            }
+            Assert.assertEquals(source.available(), body.length - Math.max(1, readSize));
+        }
+    }
+
+    @Test
+    public void shouldDetectExceptionAcrossReadBoundaries() throws Exception {
+        byte[] resultPrefix = "result-data".getBytes(StandardCharsets.UTF_8);
+        byte[] body = responseBody(resultPrefix, exceptionFrame(EXCEPTION_TAG));
+        InputStream fragmentedSource = new FilterInputStream(new ByteArrayInputStream(body)) {
+            @Override
+            public int read(byte[] buffer, int offset, int length) throws IOException {
+                return super.read(buffer, offset, Math.min(length, 1));
+            }
+        };
+
+        try (InputStream input = new HttpExceptionInputStream(fragmentedSource, EXCEPTION_TAG, 200, "query-id")) {
+            byte[] actualPrefix = new byte[resultPrefix.length];
+            int offset = 0;
+            while (offset < actualPrefix.length) {
+                int read = input.read(actualPrefix, offset, actualPrefix.length - offset);
+                Assert.assertTrue(read > 0);
+                offset += read;
+            }
+            Assert.assertEquals(actualPrefix, resultPrefix);
+
+            ServerException exception = Assert.expectThrows(ServerException.class, input::read);
+            Assert.assertEquals(exception.getCode(), 159);
+            Assert.assertEquals(exception.getQueryId(), "query-id");
+        }
+    }
+
+    @Test
+    public void shouldIgnoreFrameWithMismatchedTag() throws Exception {
+        byte[] resultPrefix = "result-data".getBytes(StandardCharsets.UTF_8);
+        byte[] body = responseBody(resultPrefix, exceptionFrame("fedcba9876543210"));
+
+        try (InputStream input = new HttpExceptionInputStream(
+                new ByteArrayInputStream(body), EXCEPTION_TAG, 200, "query-id")) {
+            Assert.assertEquals(readAll(input), body);
+        }
+    }
+
+    @Test
+    public void shouldPreserveServerExceptionWhenFrameReadFails() throws Exception {
+        byte[] resultPrefix = "result-data".getBytes(StandardCharsets.UTF_8);
+        byte[] frame = exceptionFrame(EXCEPTION_TAG);
+        byte[] body = responseBody(resultPrefix, Arrays.copyOf(frame, frame.length - 4));
+        InputStream failingSource = new FilterInputStream(new ByteArrayInputStream(body)) {
+            @Override
+            public int read(byte[] buffer, int offset, int length) throws IOException {
+                int read = super.read(buffer, offset, length);
+                if (read < 0) {
+                    throw new IOException("truncated response");
+                }
+                return read;
+            }
+        };
+
+        try (InputStream input = new HttpExceptionInputStream(failingSource, EXCEPTION_TAG, 200, "query-id")) {
+            byte[] actualPrefix = new byte[resultPrefix.length];
+            int offset = 0;
+            while (offset < actualPrefix.length) {
+                int read = input.read(actualPrefix, offset, actualPrefix.length - offset);
+                Assert.assertTrue(read > 0);
+                offset += read;
+            }
+            Assert.assertEquals(actualPrefix, resultPrefix);
+
+            ClientException exception = Assert.expectThrows(ClientException.class, input::read);
+            Assert.assertTrue(exception.getCause() instanceof ServerException);
+            Assert.assertEquals(((ServerException) exception.getCause()).getCode(), 159);
+            Assert.assertEquals(exception.getSuppressed().length, 1);
+            Assert.assertEquals(exception.getSuppressed()[0].getMessage(), "truncated response");
+        }
+    }
+
+    @DataProvider(name = "invalidFrames")
+    public Object[][] invalidFrames() {
+        String valid = new String(exceptionFrame(EXCEPTION_TAG), StandardCharsets.UTF_8);
+        return new Object[][] {
+                {valid.substring(0, valid.length() - 4)},
+                {valid.replace(ERROR_MESSAGE.length() + " " + EXCEPTION_TAG, "1 " + EXCEPTION_TAG)},
+                {valid.replace(ERROR_MESSAGE.length() + " " + EXCEPTION_TAG,
+                        ERROR_MESSAGE.length() + " fedcba9876543210")}
+        };
+    }
+
+    @Test(dataProvider = "invalidFrames")
+    public void shouldRejectIncompleteOrInvalidFrameAtEof(String frame) throws Exception {
+        try (InputStream input = new HttpExceptionInputStream(new ByteArrayInputStream(
+                frame.getBytes(StandardCharsets.UTF_8)), EXCEPTION_TAG, 200, "query-id")) {
+            ClientException exception = Assert.expectThrows(ClientException.class, input::read);
+            Assert.assertEquals(exception.getMessage(), "Incomplete ClickHouse exception frame");
+        }
+    }
+
+    private static byte[] responseBody(byte[] resultPrefix, byte[] exceptionFrame) throws IOException {
+        ByteArrayOutputStream body = new ByteArrayOutputStream();
+        body.write(resultPrefix);
+        body.write(exceptionFrame);
+        return body.toByteArray();
+    }
+
+    private static byte[] exceptionFrame(String tag) {
+        String frame = "\r\n__exception__\r\n" + tag + "\r\n" + ERROR_MESSAGE
+                + ERROR_MESSAGE.getBytes(StandardCharsets.UTF_8).length + " " + tag
+                + "\r\n__exception__\r\n";
+        return frame.getBytes(StandardCharsets.UTF_8);
+    }
+
+    private static byte[] readAll(InputStream input) throws IOException {
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+        byte[] buffer = new byte[32];
+        int read;
+        while ((read = input.read(buffer)) >= 0) {
+            output.write(buffer, 0, read);
+        }
+        return output.toByteArray();
+    }
+}
diff --git a/docs/features.md b/docs/features.md
index 0293bbc38..dc2c0d8f1 100644
--- a/docs/features.md
+++ b/docs/features.md
@@ -13,7 +13,7 @@ This document lists stable, user-visible behavior in `client-v2` and `jdbc-v2` t
 - Runtime credential updates: Existing `Client` instances can update username/password or bearer-token credentials for subsequent requests without rebuilding the client.
 - Proxy support: Can send requests through configured HTTP proxies, including proxy credentials.
 - Connection and socket tuning: Exposes pool sizing, keep-alive, reuse strategy, connect/request/socket timeouts, and low-level socket options.
-- Query execution: Executes SQL asynchronously and returns streaming query responses with response metadata and metrics.
+- Query execution: Executes SQL asynchronously and returns streaming query responses with response metadata and metrics. When a successful HTTP response ends with a tagged ClickHouse exception frame, the stream validates the frame against the `X-ClickHouse-Exception-Tag` response header and throws `ServerException` when the caller reaches it instead of exposing the frame as result data. Normal reads do not prefetch beyond the requested length unless needed to disambiguate a possible exception marker. A complete exception frame is validated by message byte length and tag without waiting for HTTP EOF; incomplete frames remain client read failures. Binary readers deliver already-decoded rows before reporting a prefetch error on the next advance.
 - Query settings: Supports per-query database selection, output format, execution limits, roles, log comments, headers, reusable `Session` objects, session settings, server settings, and network timeout overrides. Settings explicitly set to `null` or empty string will not be sent to the server. The default format (`RowBinaryWithNamesAndTypes`) is configured at the client level (`ClientConfigProperties.INPUT_OUTPUT_FORMAT`), sending the `X-ClickHouse-Format` request header. ClickHouse `26.8+` uses that request header in preference to a `FORMAT` clause in the SQL query. Configuring `format=` (empty) or `null` omits the format request header, allowing explicit SQL `FORMAT` clauses to take effect; however, on a JDBC connection this is an expert setting, as queries without a `FORMAT` clause fall back to server `default_format` (`TabSeparated`), which JDBC rejects.
 - Parameterized SQL: Accepts named query parameters and can send them through supported HTTP request encodings.
 - Result materialization helpers: Provides streaming `Records`, generic row access, and convenience APIs that materialize all rows into generic records or typed POJOs. A `Nullable(T)` column may be bound to a POJO field of a primitive type (e.g. `long`), which reads any non-`NULL` value without boxing; a value that is actually `NULL` cannot be held by such a field and is reported with a `NullValueException` (use the boxed type to accept `NULL`).
@@ -83,7 +83,7 @@ Compatibility-sensitive traits:
 - Prepared statements: Supports `?` parameters through client-side SQL rendering and validates that all parameters are bound before execution.
 - SQL parsing and classification: Classifies SQL to distinguish queries, updates, inserts, `USE`, and role-changing statements, with selectable parser backends.
 - JDBC escape processing: Translates supported JDBC escape syntax for dates, timestamps, and functions before execution. Escape sequences are only recognized outside of quoted text, so string literals and quoted identifiers — including inlined parameter values that contain `{fn `, `{d '...'}`, or `{ts '...'}` — are passed through unchanged.
-- Result set streaming: Streams result sets from ClickHouse binary formats and `FORMAT JSONEachRow`, enforces max-row limits, and manages result-set lifecycle correctly.
+- Result set streaming: Streams result sets from ClickHouse binary formats and `FORMAT JSONEachRow`, enforces max-row limits, and manages result-set lifecycle correctly. A server-side `TIMEOUT_EXCEEDED` (code 159) encountered while consuming a result set is reported as `SQLTimeoutException` with SQLState `HYT00`, including when the `ServerException` is nested in a client read failure.
 - Binary string reads: `ResultSet#getBytes(int|String)` and `ResultSet#getBinaryStream(int|String)` return the raw bytes of a `String`/`FixedString` column. Combined with the `binary_string_support` connection property, non-UTF-8/binary content stored in `String` columns round-trips byte-for-byte; `NULL` values report `null` with `wasNull()` set. `ResultSet#getObject(...)` never exposes the internal `StringValue` holder for these columns: `getObject(column, byte[].class)` returns the raw bytes, while `getObject(column, Object.class)` and the no-type `getObject(column)` overloads return a decoded `String`.
 - Result-set metadata: Exposes JDBC `ResultSetMetaData` backed by ClickHouse column schema.
 - Database metadata: Implements JDBC `DatabaseMetaData` for ClickHouse catalogs, schemas, tables, columns, and related capability reporting.
diff --git a/jdbc-v2/src/main/java/com/clickhouse/jdbc/internal/ExceptionUtils.java b/jdbc-v2/src/main/java/com/clickhouse/jdbc/internal/ExceptionUtils.java
index 81ebb4243..cd02bc646 100644
--- a/jdbc-v2/src/main/java/com/clickhouse/jdbc/internal/ExceptionUtils.java
+++ b/jdbc-v2/src/main/java/com/clickhouse/jdbc/internal/ExceptionUtils.java
@@ -8,6 +8,7 @@
 import java.net.MalformedURLException;
 import java.sql.SQLDataException;
 import java.sql.SQLException;
+import java.sql.SQLTimeoutException;
 
 /**
  * Helper class for building {@link SQLException}.
@@ -15,6 +16,7 @@
 public final class ExceptionUtils {
     public static final String SQL_STATE_CLIENT_ERROR = "HY000";
     public static final String SQL_STATE_OPERATION_CANCELLED = "HY008";
+    public static final String SQL_STATE_TIMEOUT = "HYT00";
     public static final String SQL_STATE_CONNECTION_EXCEPTION = "08000";
     public static final String SQL_STATE_SQL_ERROR = "07000";
     public static final String SQL_STATE_NO_DATA = "02000";
@@ -28,6 +30,8 @@ public final class ExceptionUtils {
     public static final String SQL_STATE_WRONG_OBJECT_TYPE = "42809";
     public static final String SQL_STATE_TYPE_MISMATCH = "2200G";
 
+    private static final int CLICKHOUSE_TIMEOUT_EXCEEDED = 159;
+
     private ExceptionUtils() {}//Private constructor
 
     // https://en.wikipedia.org/wiki/SQLSTATE
@@ -56,6 +60,12 @@ public static SQLException toSqlState(String message, String debugMessage, Excep
 
         if (cause instanceof SQLException) {
             return (SQLException) cause;
+        }
+
+        ServerException serverException = findServerException(cause);
+        if (serverException != null && serverException.getCode() == CLICKHOUSE_TIMEOUT_EXCEEDED) {
+            String timeoutMessage = message == null ? serverException.getMessage() : message;
+            return new SQLTimeoutException(timeoutMessage, SQL_STATE_TIMEOUT, serverException.getCode(), cause);
         } else if (cause instanceof ClientMisconfigurationException) {
             return new SQLException(exceptionMessage, SQL_STATE_CLIENT_ERROR, cause);
         } else if (cause instanceof ConnectionInitiationException) {
@@ -75,6 +85,18 @@ public static SQLException toSqlState(String message, String debugMessage, Excep
         return new SQLException(exceptionMessage, SQL_STATE_CLIENT_ERROR, cause);//Default
     }
 
+    private static ServerException findServerException(Throwable throwable) {
+        for (Throwable cause = throwable; cause != null; cause = cause.getCause()) {
+            if (cause instanceof ServerException) {
+                return (ServerException) cause;
+            }
+            if (cause.getCause() == cause) {
+                break;
+            }
+        }
+        return null;
+    }
+
     public static Throwable getRootCause(Throwable throwable) {
         Throwable cause = throwable;
         while (cause.getCause() != null) {
diff --git a/jdbc-v2/src/test/java/com/clickhouse/jdbc/ResultSetTimeoutTest.java b/jdbc-v2/src/test/java/com/clickhouse/jdbc/ResultSetTimeoutTest.java
new file mode 100644
index 000000000..f659561a6
--- /dev/null
+++ b/jdbc-v2/src/test/java/com/clickhouse/jdbc/ResultSetTimeoutTest.java
@@ -0,0 +1,84 @@
+package com.clickhouse.jdbc;
+
+import com.clickhouse.client.api.ClientConfigProperties;
+import com.clickhouse.client.api.ServerException;
+import com.clickhouse.client.api.http.ClickHouseHttpProto;
+import com.clickhouse.jdbc.internal.ExceptionUtils;
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.client.WireMock;
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
+import org.apache.hc.core5.http.HttpStatus;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import org.testng.annotations.DataProvider;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLTimeoutException;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Properties;
+
+public class ResultSetTimeoutTest {
+
+    @DataProvider(name = "serverErrors")
+    public Object[][] serverErrors() {
+        return new Object[][] {{159, "HYT00", SQLTimeoutException.class}, {60, "42S02", SQLException.class}};
+    }
+
+    @Test(dataProvider = "serverErrors")
+    public void shouldDeliverAllRowsBeforeServerError(int code, String sqlState,
+                                                     Class<? extends SQLException> exceptionClass) throws Exception {
+        String exceptionTag = "0123456789abcdef";
+        String queryId = "result-set-timeout";
+        String errorMessage = "Code: " + code + ". DB::Exception: Query failed.\n";
+        String exceptionFrame = "\r\n__exception__\r\n" + exceptionTag + "\r\n" + errorMessage
+                + errorMessage.getBytes(StandardCharsets.UTF_8).length + " " + exceptionTag
+                + "\r\n__exception__\r\n";
+        // RowBinaryWithNamesAndTypes schema followed by two rows; next() prefetches one row ahead.
+        byte[] rowBinaryResultPrefix = {
+                0x01, 0x01, 0x31, 0x05, 0x55, 0x49, 0x6e, 0x74, 0x38, 0x01, 0x02
+        };
+        ByteArrayOutputStream body = new ByteArrayOutputStream();
+        body.write(rowBinaryResultPrefix);
+        body.write(exceptionFrame.getBytes(StandardCharsets.UTF_8));
+
+        WireMockServer mockServer = new WireMockServer(WireMockConfiguration.options().dynamicPort());
+        mockServer.start();
+
+        try {
+            mockServer.stubFor(WireMock.post(WireMock.anyUrl())
+                    .willReturn(WireMock.aResponse()
+                            .withStatus(HttpStatus.SC_OK)
+                            .withHeader(ClickHouseHttpProto.HEADER_EXCEPTION_TAG, exceptionTag)
+                            .withHeader(ClickHouseHttpProto.HEADER_QUERY_ID, queryId)
+                            .withBody(body.toByteArray())));
+
+            Properties properties = new Properties();
+            properties.setProperty(ClientConfigProperties.SERVER_TIMEZONE.getKey(), "UTC");
+            properties.setProperty(ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getKey(), "false");
+            properties.setProperty(ClientConfigProperties.USE_HTTP_COMPRESSION.getKey(), "false");
+
+            String jdbcUrl = "jdbc:clickhouse://localhost:" + mockServer.port() + "/default";
+            try (Connection connection = new ConnectionImpl(jdbcUrl, properties);
+                 Statement statement = connection.createStatement();
+                 ResultSet resultSet = statement.executeQuery("SELECT 1")) {
+                Assert.assertTrue(resultSet.next());
+                Assert.assertEquals(resultSet.getInt(1), 1);
+                Assert.assertTrue(resultSet.next());
+                Assert.assertEquals(resultSet.getInt(1), 2);
+
+                SQLException exception = Assert.expectThrows(exceptionClass, resultSet::next);
+                Assert.assertEquals(exception.getErrorCode(), code);
+                Assert.assertEquals(exception.getSQLState(), sqlState);
+                Assert.assertTrue(ExceptionUtils.getRootCause(exception) instanceof ServerException);
+                Assert.assertEquals(((ServerException) ExceptionUtils.getRootCause(exception)).getCode(), code);
+                Assert.assertTrue(exception.getMessage().startsWith(errorMessage.trim()), exception.getMessage());
+            }
+        } finally {
+            mockServer.stop();
+        }
+    }
+}