diff --git a/presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java b/presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java
index 95925891a..c1ed1e51a 100644
--- a/presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java
+++ b/presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java
@@ -33,6 +33,7 @@ import io.prestosql.failuredetector.FailureDetector;
 import io.prestosql.metadata.InternalNode;
 import io.prestosql.metadata.Split;
 import io.prestosql.operator.HttpPageBufferClient;
+import io.prestosql.server.remotetask.SimpleHttpResponseHandler;
 import io.prestosql.snapshot.QuerySnapshotManager;
 import io.prestosql.spi.PrestoException;
 import io.prestosql.spi.QueryId;
@@ -610,6 +611,12 @@ public final class SqlStageExecution
                             return;
                         }
                     }
+                    else if (message.contains(SimpleHttpResponseHandler.EXPECT_200_SAW_5XX)) {
+                        // SimpleHttpResponseHandler can also produce errors that are resumable
+                        log.debug(failure, "Task %s on node %s failed but is resumable. Triggering rescheduling.", taskStatus.getTaskId(), taskStatus.getNodeId());
+                        stateMachine.transitionToResumableFailure();
+                        return;
+                    }
                 }
                 stateMachine.transitionToFailed(failure);
             }
diff --git a/presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java b/presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java
index dccd1061b..177dce407 100644
--- a/presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java
+++ b/presto-main/src/main/java/io/prestosql/operator/ExchangeClient.java
@@ -88,9 +88,11 @@ public class ExchangeClient
     private boolean noMoreTargets;
     private final Set<String> allTargets = new HashSet<>();
     // Markers received before all targets are known. These markers will be sent to all new targets.
-    private final List<SerializedPage> pendingMarkers = Collections.synchronizedList(new ArrayList<>());
+    @GuardedBy("this")
+    private final List<SerializedPage> pendingMarkers = new ArrayList<>();
     // pendingOrigins keeps track of the origins in pendingMarkers
-    private final List<Optional<String>> pendingOrigins = Collections.synchronizedList(new ArrayList<>());
+    @GuardedBy("this")
+    private final List<Optional<String>> pendingOrigins = new ArrayList<>();
 
     @GuardedBy("this")
     private final Deque<HttpPageBufferClient> queuedClients = new LinkedList<>();
@@ -209,7 +211,7 @@ public class ExchangeClient
         }
     }
 
-    public void noMoreTargets()
+    public synchronized void noMoreTargets()
     {
         noMoreTargets = true;
         pendingMarkers.clear();
diff --git a/presto-main/src/main/java/io/prestosql/server/remotetask/HttpRemoteTask.java b/presto-main/src/main/java/io/prestosql/server/remotetask/HttpRemoteTask.java
index 60e221d87..052770735 100644
--- a/presto-main/src/main/java/io/prestosql/server/remotetask/HttpRemoteTask.java
+++ b/presto-main/src/main/java/io/prestosql/server/remotetask/HttpRemoteTask.java
@@ -103,6 +103,7 @@ import static io.prestosql.protocol.JsonCodecWrapper.unwrapJsonCodec;
 import static io.prestosql.protocol.RequestHelpers.setContentTypeHeaders;
 import static io.prestosql.server.remotetask.RequestErrorTracker.logError;
 import static io.prestosql.spi.StandardErrorCode.REMOTE_HOST_GONE;
+import static io.prestosql.spi.StandardErrorCode.REMOTE_TASK_ERROR;
 import static io.prestosql.spi.StandardErrorCode.REMOTE_TASK_MISMATCH;
 import static io.prestosql.spi.StandardErrorCode.TOO_MANY_REQUESTS_FAILED;
 import static io.prestosql.util.Failures.WORKER_NODE_ERROR;
@@ -831,6 +832,7 @@ public final class HttpRemoteTask
         return failureInfo.getErrorCode().equals(TOO_MANY_REQUESTS_FAILED.toErrorCode())
                 || failureInfo.getErrorCode().equals(REMOTE_HOST_GONE.toErrorCode())
                 || failureInfo.getErrorCode().equals(REMOTE_TASK_MISMATCH.toErrorCode())
+                || failureInfo.getErrorCode().equals(REMOTE_TASK_ERROR.toErrorCode())
                 || failureInfo.getMessage() != null && failureInfo.getMessage().contains(WORKER_NODE_ERROR);
     }
 
diff --git a/presto-main/src/main/java/io/prestosql/server/remotetask/SimpleHttpResponseHandler.java b/presto-main/src/main/java/io/prestosql/server/remotetask/SimpleHttpResponseHandler.java
index 0d67e1c5d..6f0280c21 100644
--- a/presto-main/src/main/java/io/prestosql/server/remotetask/SimpleHttpResponseHandler.java
+++ b/presto-main/src/main/java/io/prestosql/server/remotetask/SimpleHttpResponseHandler.java
@@ -30,6 +30,8 @@ import static java.util.Objects.requireNonNull;
 public class SimpleHttpResponseHandler<T>
         implements FutureCallback<BaseResponse<T>>
 {
+    // Used by SqlStageExecution to detect 5xx failures
+    public static final CharSequence EXPECT_200_SAW_5XX = format("to be %d, but was 5", HttpStatus.OK.code());
     private final SimpleHttpResponseCallback<T> callback;
 
     private final URI uri;
@@ -62,6 +64,7 @@ public class SimpleHttpResponseHandler<T>
                         cause = new PrestoException(REMOTE_TASK_ERROR, format("Expected response from %s is empty", uri));
                     }
                     else {
+                        // CAUTION: keep error message consistent with EXPECT_200_SAW_5XX.
                         cause = new PrestoException(REMOTE_TASK_ERROR, format("Expected response code from %s to be %s, but was %s%n%s",
                                 uri,
                                 HttpStatus.OK.code(),
@@ -84,12 +87,14 @@ public class SimpleHttpResponseHandler<T>
     private String createErrorMessage(BaseResponse<T> response)
     {
         if (response instanceof JsonResponseWrapper) {
+            // CAUTION: keep error message consistent with EXPECT_200_SAW_5XX.
             return format("Expected response code from %s to be %s, but was %s%n%s",
                     uri,
                     OK.code(),
                     response.getStatusCode(),
                     unwrapJsonResponse(response).getResponseBody());
         }
+        // CAUTION: keep error message consistent with EXPECT_200_SAW_5XX.
         return format("Expected response code from %s to be %s, but was %s: %s",
                 uri,
                 OK.code(),
diff --git a/presto-spi/src/main/java/io/prestosql/spi/snapshot/MarkerPage.java b/presto-spi/src/main/java/io/prestosql/spi/snapshot/MarkerPage.java
index f081cc181..a4ebc61ac 100644
--- a/presto-spi/src/main/java/io/prestosql/spi/snapshot/MarkerPage.java
+++ b/presto-spi/src/main/java/io/prestosql/spi/snapshot/MarkerPage.java
@@ -14,7 +14,6 @@
  */
 package io.prestosql.spi.snapshot;
 
-import com.google.common.primitives.Ints;
 import com.google.common.primitives.Longs;
 import io.prestosql.spi.Page;
 
@@ -90,7 +89,7 @@ public class MarkerPage
 
     public byte[] serialize()
     {
-        final int size = Longs.BYTES + Ints.BYTES + 1;
+        final int size = Longs.BYTES + 1;
         return ByteBuffer.allocate(size)
                 .putLong(snapshotId)
                 .put((byte) (isResuming ? 1 : 0))