diff --git a/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java b/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java
index b9e6092c7..4c28a4740 100644
--- a/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java
+++ b/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java
@@ -431,7 +431,7 @@ public final class HttpPageBufferClient
                 Throwable throwable = rewriteException(t);
                 boolean fail = failureRetryPolicy.hasFailed(fromUri(uri));
 
-                if (!(throwable instanceof PrestoException) && fail) {
+                if ((!(throwable instanceof PrestoException) || (throwable instanceof PageTransportServerException)) && fail) {
                     String message = format("%s (%s - %s failures, failure duration %s, total failed request time %s)",
                             WORKER_NODE_ERROR,
                             uri,
@@ -447,7 +447,7 @@ public final class HttpPageBufferClient
                     }
                     throwable = new PageTransportTimeoutException(fromUri(uri), message, throwable);
                 }
-                handleFailure(throwable, resultFuture);
+                handleFailure(throwable, resultFuture, fail);
             }
         }, pageBufferClientCallbackExecutor);
     }
@@ -505,7 +505,7 @@ public final class HttpPageBufferClient
         checkState(!Thread.holdsLock(lock), "Cannot execute this method while holding a lock");
     }
 
-    private void handleFailure(Throwable t, HttpResponseFuture<?> expectedFuture)
+    private void handleFailure(Throwable t, HttpResponseFuture<?> expectedFuture, boolean failed)
     {
         // Can not delegate to other callback while holding a lock on this
         checkNotHoldsLock(this);
@@ -513,7 +513,8 @@ public final class HttpPageBufferClient
         requestsFailed.incrementAndGet();
         requestsCompleted.incrementAndGet();
 
-        if (t instanceof PrestoException) {
+        if ((t instanceof PageTransportServerException && failed) // resumable failure, failed even after retries
+                || (!(t instanceof PageTransportServerException)) && t instanceof PrestoException) {
             clientCallback.clientFailed(HttpPageBufferClient.this, t);
         }
 
@@ -526,6 +527,11 @@ public final class HttpPageBufferClient
         clientCallback.requestComplete(HttpPageBufferClient.this);
     }
 
+    private void handleFailure(Throwable t, HttpResponseFuture<?> expectedFuture)
+    {
+        handleFailure(t, expectedFuture, true);
+    }
+
     @Override
     public boolean equals(Object o)
     {
@@ -595,7 +601,7 @@ public final class HttpPageBufferClient
         {
             if (exception instanceof RuntimeException || exception instanceof IOException) {
                 throw propagate(request,
-                        new PageTransportErrorException(format("%s %d! Error fetching %s: %s",
+                        new PageTransportServerException(format("%s %d! Error fetching %s: %s",
                                 PAGE_TRANSPORT_ERROR_PREFIX, HttpStatus.INTERNAL_SERVER_ERROR.code(),
                                 request.getUri().toASCIIString(), exception.getMessage()),
                         exception));
diff --git a/presto-main/src/main/java/io/prestosql/operator/PageTransportServerException.java b/presto-main/src/main/java/io/prestosql/operator/PageTransportServerException.java
new file mode 100644
index 000000000..e8fc37e00
--- /dev/null
+++ b/presto-main/src/main/java/io/prestosql/operator/PageTransportServerException.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.prestosql.operator;
+
+import io.prestosql.spi.PrestoException;
+
+import static io.prestosql.spi.StandardErrorCode.PAGE_TRANSPORT_ERROR;
+
+public class PageTransportServerException
+        extends PrestoException
+{
+    public PageTransportServerException(String message)
+    {
+        super(PAGE_TRANSPORT_ERROR, message);
+    }
+
+    public PageTransportServerException(String message, Throwable cause)
+    {
+        super(PAGE_TRANSPORT_ERROR, message, cause);
+    }
+}
diff --git a/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java b/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java
index 6e77fd0b1..44af4b276 100644
--- a/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java
+++ b/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java
@@ -298,6 +298,45 @@ public class TestHttpPageBufferClient
         assertStatus(client, location, "closed", 0, 1, 2, 1, "not scheduled");
     }
 
+    @Test
+    public void testResumableServerFailures()
+            throws Exception
+    {
+        TestingTicker ticker = new TestingTicker();
+        AtomicReference<Duration> tickerIncrement = new AtomicReference<>(new Duration(0, TimeUnit.SECONDS));
+
+        TestingHttpClient.Processor processor = (input) -> {
+            Duration delta = tickerIncrement.get();
+            ticker.increment(delta.toMillis(), TimeUnit.MILLISECONDS);
+            throw new PageTransportServerException("Foo");
+        };
+
+        CyclicBarrier requestComplete = new CyclicBarrier(2);
+        TestingClientCallback callback = new TestingClientCallback(requestComplete);
+
+        URI location = URI.create("http://localhost:8080");
+        String instanceId = "testing instance id";
+        HttpPageBufferClient client = new HttpPageBufferClient(new TestingHttpClient(processor, scheduler),
+                new DataSize(10, Unit.MEGABYTE),
+                true,
+                new TaskLocation(location, instanceId),
+                callback,
+                scheduler,
+                pageBufferClientCallbackExecutor,
+                false,
+                null, failureDetectorManager);
+
+        assertStatus(client, location, "queued", 0, 0, 0, 0, "not scheduled");
+
+        client.scheduleRequest();
+        requestComplete.await(10, TimeUnit.SECONDS);
+        assertEquals(callback.getPages().size(), 0);
+        assertEquals(callback.getCompletedRequests(), 1);
+        assertEquals(callback.getFinishedBuffers(), 0);
+        assertEquals(callback.getFailedBuffers(), 0);
+        assertStatus(client, location, "queued", 0, 1, 1, 1, "not scheduled");
+    }
+
     @Test
     public void testInvalidResponses()
             throws Exception