diff --git a/tez-api/src/main/java/org/apache/tez/runtime/api/events/InputReadErrorEvent.java b/tez-api/src/main/java/org/apache/tez/runtime/api/events/InputReadErrorEvent.java
index cabc39f..8ef50eb 100644
--- a/tez-api/src/main/java/org/apache/tez/runtime/api/events/InputReadErrorEvent.java
+++ b/tez-api/src/main/java/org/apache/tez/runtime/api/events/InputReadErrorEvent.java
@@ -51,26 +51,44 @@
    */
   private final int numFailures;
 
-  private InputReadErrorEvent(final String diagnostics, final int index,
-                              final int version, final int numFailures) {
+  /**
+   * Whether this input read error is caused while fetching local file.
+   */
+  private final boolean isLocalFetch;
+
+  /**
+   * Whether this input read error is caused because the fetcher detected a fatal, unrecoverable,
+   * local file read issue from the shuffle handler.
+   */
+  private final boolean isDiskErrorAtSource;
+
+  private InputReadErrorEvent(final String diagnostics, final int index, final int version,
+      final int numFailures, boolean isLocalFetch, boolean isDiskErrorAtSource) {
     super();
     this.diagnostics = diagnostics;
     this.index = index;
     this.version = version;
     this.numFailures = numFailures;
+    this.isLocalFetch = isLocalFetch;
+    this.isDiskErrorAtSource = isDiskErrorAtSource;
   }
 
-  public static InputReadErrorEvent create(String diagnostics, int index,
-                                           int version) {
-    return create(diagnostics, index, version, 1);
+  public static InputReadErrorEvent create(String diagnostics, int index, int version,
+      boolean isLocalFetch, boolean isDiskErrorAtSource) {
+    return create(diagnostics, index, version, 1, isLocalFetch, isDiskErrorAtSource);
+  }
+
+  public static InputReadErrorEvent create(String diagnostics, int index, int version) {
+    return create(diagnostics, index, version, 1, false, false);
   }
 
   /**
    * Create an InputReadErrorEvent.
    */
   public static InputReadErrorEvent create(final String diagnostics, final int index,
-      final int version, final int numFailures) {
-    return new InputReadErrorEvent(diagnostics, index, version, numFailures);
+      final int version, final int numFailures, boolean isLocalFetch, boolean isDiskErrorAtSource) {
+    return new InputReadErrorEvent(diagnostics, index, version, numFailures, isLocalFetch,
+        isDiskErrorAtSource);
   }
 
   public String getDiagnostics() {
@@ -92,6 +110,14 @@
     return numFailures;
   }
 
+  public boolean isLocalFetch() {
+    return isLocalFetch;
+  }
+
+  public boolean isDiskErrorAtSource() {
+    return isDiskErrorAtSource;
+  }
+
   @Override
   public int hashCode() {
     return Objects.hash(index, version);
diff --git a/tez-api/src/main/java/org/apache/tez/runtime/library/common/shuffle/api/ShuffleHandlerError.java b/tez-api/src/main/java/org/apache/tez/runtime/library/common/shuffle/api/ShuffleHandlerError.java
new file mode 100644
index 0000000..09137de
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/runtime/library/common/shuffle/api/ShuffleHandlerError.java
@@ -0,0 +1,27 @@
+/**
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you 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 org.apache.tez.runtime.library.common.shuffle.api;
+
+/**
+ * ShuffleHandlerError enum encapsulates possible error messages that can be propagated from
+ * ShuffleHandler to fetchers. Depending on the message, fetchers can make better decisions, or give
+ * AM a hint in order to let it make better decisions in case of shuffle issues.
+ */
+public enum ShuffleHandlerError {
+  DISK_ERROR_EXCEPTION
+}
diff --git a/tez-api/src/main/java/org/apache/tez/runtime/library/common/shuffle/api/package-info.java b/tez-api/src/main/java/org/apache/tez/runtime/library/common/shuffle/api/package-info.java
new file mode 100644
index 0000000..9ad8e61
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/runtime/library/common/shuffle/api/package-info.java
@@ -0,0 +1,22 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+@Private
+package org.apache.tez.runtime.library.common.shuffle.api;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
\ No newline at end of file
diff --git a/tez-api/src/main/proto/Events.proto b/tez-api/src/main/proto/Events.proto
index 7123500..e041c33 100644
--- a/tez-api/src/main/proto/Events.proto
+++ b/tez-api/src/main/proto/Events.proto
@@ -39,6 +39,8 @@
   optional int32 index = 1;
   optional string diagnostics = 2;
   optional int32 version = 3;
+  optional bool is_local_fetch = 4;
+  optional bool is_disk_error_at_source = 5;
 }
 
 message InputFailedEventProto {
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
index 37e818e..9a5e73d 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
@@ -1825,6 +1825,7 @@
             + " at inputIndex " + failedInputIndexOnDestTa);
       long time = attempt.clock.getTime();
       Long firstErrReportTime = attempt.uniquefailedOutputReports.get(failedDestTaId);
+
       if (firstErrReportTime == null) {
         attempt.uniquefailedOutputReports.put(failedDestTaId, time);
         firstErrReportTime = time;
@@ -1851,7 +1852,8 @@
       // If needed we can launch a background task without failing this task
       // to generate a copy of the output just in case.
       // If needed we can consider only running consumer tasks
-      if (!crossTimeDeadline && withinFailureFractionLimits && withinOutputFailureLimits) {
+      if (!crossTimeDeadline && withinFailureFractionLimits && withinOutputFailureLimits
+          && !(readErrorEvent.isLocalFetch() || readErrorEvent.isDiskErrorAtSource())) {
         return attempt.getInternalState();
       }
       String message = attempt.getID() + " being failed for too many output errors. "
@@ -1862,7 +1864,10 @@
           + ", MAX_ALLOWED_OUTPUT_FAILURES=" + maxAllowedOutputFailures
           + ", MAX_ALLOWED_TIME_FOR_TASK_READ_ERROR_SEC="
           + maxAllowedTimeForTaskReadErrorSec
-          + ", readErrorTimespan=" + readErrorTimespanSec;
+          + ", readErrorTimespan=" + readErrorTimespanSec
+          + ", isLocalFetch=" + readErrorEvent.isLocalFetch()
+          + ", isDiskErrorAtSource=" + readErrorEvent.isDiskErrorAtSource();
+
       LOG.info(message);
       attempt.addDiagnosticInfo(message);
       // send input failed event
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskAttempt.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskAttempt.java
index 41cce3b..6862bec 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskAttempt.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskAttempt.java
@@ -19,7 +19,6 @@
 package org.apache.tez.dag.app.dag.impl;
 
 import org.apache.tez.dag.app.MockClock;
-import org.apache.tez.dag.app.rm.AMSchedulerEventTAStateUpdated;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;
@@ -83,6 +82,7 @@
 import org.apache.tez.dag.app.TaskCommunicatorWrapper;
 import org.apache.tez.dag.app.TaskHeartbeatHandler;
 import org.apache.tez.dag.app.dag.TaskAttemptStateInternal;
+import org.apache.tez.dag.app.dag.DAG;
 import org.apache.tez.dag.app.dag.Task;
 import org.apache.tez.dag.app.dag.Vertex;
 import org.apache.tez.dag.app.dag.event.DAGEvent;
@@ -127,6 +127,7 @@
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -2154,6 +2155,52 @@
     Assert.assertEquals(1, taImpl.taskAttemptFinishedEventLogged);
   }
 
+  @Test
+  public void testMapTaskIsBlamedImmediatelyOnLocalFetchFailure() throws ServicePluginException {
+    // local fetch failure or disk read error at source -> turn source attempt to FAIL_IN_PROGRESS
+    testMapTaskFailingForFetchFailureType(true, true, TaskAttemptStateInternal.FAIL_IN_PROGRESS);
+    testMapTaskFailingForFetchFailureType(true, false, TaskAttemptStateInternal.FAIL_IN_PROGRESS);
+    testMapTaskFailingForFetchFailureType(false, true, TaskAttemptStateInternal.FAIL_IN_PROGRESS);
+
+    // remote fetch failure -> won't change current state
+    testMapTaskFailingForFetchFailureType(false, false, TaskAttemptStateInternal.NEW);
+  }
+
+  private void testMapTaskFailingForFetchFailureType(boolean isLocalFetch,
+      boolean isDiskErrorAtSource, TaskAttemptStateInternal expectedState) {
+    EventHandler eventHandler = mock(EventHandler.class);
+    TezTaskID taskID =
+        TezTaskID.getInstance(TezVertexID.getInstance(TezDAGID.getInstance("1", 1, 1), 1), 1);
+    TaskAttemptImpl sourceAttempt = new MockTaskAttemptImpl(taskID, 1, eventHandler, null,
+        new Configuration(), SystemClock.getInstance(), mock(TaskHeartbeatHandler.class), appCtx,
+        false, null, null, false);
+
+    // the original read error event, sent by reducer task
+    InputReadErrorEvent inputReadErrorEvent =
+        InputReadErrorEvent.create("", 0, 1, 1, isLocalFetch, isDiskErrorAtSource);
+    TezTaskAttemptID destTaskAttemptId = mock(TezTaskAttemptID.class);
+    when(destTaskAttemptId.getTaskID()).thenReturn(mock(TezTaskID.class));
+    when(destTaskAttemptId.getTaskID().getVertexID()).thenReturn(mock(TezVertexID.class));
+    when(appCtx.getCurrentDAG()).thenReturn(mock(DAG.class));
+    when(appCtx.getCurrentDAG().getVertex(Mockito.any(TezVertexID.class)))
+        .thenReturn(mock(Vertex.class));
+    when(appCtx.getCurrentDAG().getVertex(Mockito.any(TezVertexID.class)).getRunningTasks())
+        .thenReturn(100);
+
+    EventMetaData mockMeta = mock(EventMetaData.class);
+    when(mockMeta.getTaskAttemptID()).thenReturn(destTaskAttemptId);
+    TezEvent tezEvent = new TezEvent(inputReadErrorEvent, mockMeta);
+
+    // the event is propagated to map task's event handler
+    TaskAttemptEventOutputFailed outputFailedEvent =
+        new TaskAttemptEventOutputFailed(sourceAttempt.getID(), tezEvent, 11);
+
+    Assert.assertEquals(TaskAttemptStateInternal.NEW, sourceAttempt.getInternalState());
+    TaskAttemptStateInternal resultState = new TaskAttemptImpl.OutputReportedFailedTransition()
+        .transition(sourceAttempt, outputFailedEvent);
+    Assert.assertEquals(expectedState, resultState);
+  }
+
   private Event verifyEventType(List<Event> events,
       Class<? extends Event> eventClass, int expectedOccurences) {
     int count = 0;
diff --git a/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java b/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java
index f294edc..b67883d 100644
--- a/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java
+++ b/tez-plugins/tez-aux-services/src/main/java/org/apache/tez/auxservices/ShuffleHandler.java
@@ -71,6 +71,7 @@
 import org.apache.tez.common.security.JobTokenSecretManager;
 import org.apache.tez.runtime.library.common.Constants;
 import org.apache.tez.runtime.library.common.security.SecureShuffleUtils;
+import org.apache.tez.runtime.library.common.shuffle.api.ShuffleHandlerError;
 import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.ShuffleHeader;
 import org.apache.hadoop.metrics2.MetricsSystem;
 import org.apache.hadoop.metrics2.annotation.Metric;
@@ -84,6 +85,7 @@
 import org.apache.hadoop.security.token.Token;
 import org.apache.tez.runtime.library.common.sort.impl.TezIndexRecord;
 import org.apache.hadoop.util.Shell;
+import org.apache.hadoop.util.DiskChecker.DiskErrorException;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.proto.YarnServerCommonProtos.VersionProto;
@@ -101,6 +103,7 @@
 import org.iq80.leveldb.Logger;
 import org.iq80.leveldb.Options;
 import org.jboss.netty.bootstrap.ServerBootstrap;
+import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBuffers;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelFactory;
@@ -608,6 +611,10 @@
     return new Shuffle(conf);
   }
 
+  protected JobTokenSecretManager getSecretManager() {
+    return secretManager;
+  }
+
   private void recoverState(Configuration conf) throws IOException {
     Path recoveryRoot = getRecoveryPath();
     if (recoveryRoot != null) {
@@ -727,7 +734,7 @@
   private void addJobToken(JobID jobId, String user,
       Token<JobTokenIdentifier> jobToken) {
     userRsrc.put(jobId.toString(), user);
-    secretManager.addTokenForJob(jobId.toString(), jobToken);
+    getSecretManager().addTokenForJob(jobId.toString(), jobToken);
     LOG.info("Added token for " + jobId.toString());
   }
 
@@ -772,7 +779,7 @@
 
   private void removeJobShuffleInfo(JobID jobId) throws IOException {
     String jobIdStr = jobId.toString();
-    secretManager.removeTokenForJob(jobIdStr);
+    getSecretManager().removeTokenForJob(jobIdStr);
     userRsrc.remove(jobIdStr);
     if (stateDb != null) {
       try {
@@ -1080,11 +1087,19 @@
       try {
         populateHeaders(mapIds, jobId, dagId, user, reduceRange,
           response, keepAliveParam, mapOutputInfoMap);
-      } catch(IOException e) {
+      } catch (DiskErrorException e) { // fatal error: fetcher should be aware of that
+        LOG.error("Shuffle error in populating headers (fatal: DiskErrorException):", e);
+        String errorMessage = getErrorMessage(e);
+        // custom message, might be noticed by fetchers
+        // it should reuse the current response object, as headers have been already set for it
+        sendFakeShuffleHeaderWithError(ctx,
+            ShuffleHandlerError.DISK_ERROR_EXCEPTION + ": " + errorMessage, response);
+        return;
+      } catch (IOException e) {
         ch.write(response);
         LOG.error("Shuffle error in populating headers :", e);
         String errorMessage = getErrorMessage(e);
-        sendError(ctx,errorMessage , INTERNAL_SERVER_ERROR);
+        sendError(ctx, errorMessage, INTERNAL_SERVER_ERROR);
         return;
       }
       ch.write(response);
@@ -1337,7 +1352,7 @@
     protected void verifyRequest(String appid, ChannelHandlerContext ctx,
         HttpRequest request, HttpResponse response, URL requestUri)
         throws IOException {
-      SecretKey tokenSecret = secretManager.retrieveTokenSecret(appid);
+      SecretKey tokenSecret = getSecretManager().retrieveTokenSecret(appid);
       if (null == tokenSecret) {
         LOG.info("Request for unknown token " + appid);
         throw new IOException("could not find jobid");
@@ -1444,22 +1459,37 @@
       return writeFuture;
     }
 
-    protected void sendError(ChannelHandlerContext ctx,
-        HttpResponseStatus status) {
+    protected void sendError(ChannelHandlerContext ctx, HttpResponseStatus status) {
       sendError(ctx, "", status);
     }
 
-    protected void sendError(ChannelHandlerContext ctx, String message,
-        HttpResponseStatus status) {
+    protected void sendError(ChannelHandlerContext ctx, String message, HttpResponseStatus status) {
       HttpResponse response = new DefaultHttpResponse(HTTP_1_1, status);
+      sendError(ctx, message, response);
+    }
+
+    protected void sendError(ChannelHandlerContext ctx, String message, HttpResponse response) {
+      sendError(ctx, ChannelBuffers.copiedBuffer(message, CharsetUtil.UTF_8), response);
+    }
+
+    private void sendFakeShuffleHeaderWithError(ChannelHandlerContext ctx, String message,
+        HttpResponse response) throws IOException {
+      ShuffleHeader header = new ShuffleHeader(message, -1, -1, -1);
+      DataOutputBuffer out = new DataOutputBuffer();
+      header.write(out);
+
+      sendError(ctx, wrappedBuffer(out.getData(), 0, out.getLength()), response);
+    }
+
+    protected void sendError(ChannelHandlerContext ctx, ChannelBuffer content,
+        HttpResponse response) {
       response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");
       // Put shuffle version into http header
       response.headers().set(ShuffleHeader.HTTP_HEADER_NAME,
           ShuffleHeader.DEFAULT_HTTP_HEADER_NAME);
       response.headers().set(ShuffleHeader.HTTP_HEADER_VERSION,
           ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION);
-      response.setContent(
-        ChannelBuffers.copiedBuffer(message, CharsetUtil.UTF_8));
+      response.setContent(content);
 
       // Close the connection as soon as the error message is sent.
       ctx.getChannel().write(response).addListener(ChannelFutureListener.CLOSE);
diff --git a/tez-plugins/tez-aux-services/src/test/java/org/apache/tez/auxservices/TestShuffleHandler.java b/tez-plugins/tez-aux-services/src/test/java/org/apache/tez/auxservices/TestShuffleHandler.java
index a7f4446..5ca4ed8 100644
--- a/tez-plugins/tez-aux-services/src/test/java/org/apache/tez/auxservices/TestShuffleHandler.java
+++ b/tez-plugins/tez-aux-services/src/test/java/org/apache/tez/auxservices/TestShuffleHandler.java
@@ -20,6 +20,7 @@
 //import static org.apache.hadoop.test.MetricsAsserts.assertCounter;
 //import static org.apache.hadoop.test.MetricsAsserts.assertGauge;
 //import static org.apache.hadoop.test.MetricsAsserts.getMetrics;
+import org.apache.hadoop.util.DiskChecker.DiskErrorException;
 import static org.junit.Assert.assertTrue;
 import static org.jboss.netty.buffer.ChannelBuffers.wrappedBuffer;
 import static org.jboss.netty.handler.codec.http.HttpResponseStatus.OK;
@@ -41,15 +42,12 @@
 import java.net.URL;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.zip.CheckedOutputStream;
 import java.util.zip.Checksum;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
-import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
@@ -58,11 +56,14 @@
 import org.apache.hadoop.io.WritableUtils;
 import org.apache.hadoop.io.nativeio.NativeIO;
 import org.apache.hadoop.mapred.JobID;
-import org.apache.hadoop.mapred.MapTask;
 import org.apache.hadoop.mapreduce.TypeConverter;
 import org.apache.tez.runtime.library.common.security.SecureShuffleUtils;
 import org.apache.tez.common.security.JobTokenIdentifier;
 import org.apache.tez.common.security.JobTokenSecretManager;
+import org.apache.tez.http.BaseHttpConnection;
+import org.apache.tez.http.HttpConnectionParams;
+import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
+import org.apache.tez.runtime.library.common.shuffle.api.ShuffleHandlerError;
 import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.ShuffleHeader;
 import org.apache.hadoop.metrics2.MetricsSystem;
 import org.apache.hadoop.metrics2.impl.MetricsSystemImpl;
@@ -172,6 +173,52 @@
     }
   }
 
+  class MockShuffleHandlerWithFatalDiskError extends org.apache.tez.auxservices.ShuffleHandler {
+    public static final String MESSAGE =
+        "Could not find application_1234/240/output/attempt_1234_0/file.out.index";
+
+    private JobTokenSecretManager secretManager =
+        new JobTokenSecretManager(JobTokenSecretManager.createSecretKey(getSecret().getBytes()));
+
+    protected JobTokenSecretManager getSecretManager(){
+      return secretManager;
+    }
+
+    @Override
+    protected Shuffle getShuffle(final Configuration conf) {
+      return new Shuffle(conf) {
+        @Override
+        protected void verifyRequest(String appid, ChannelHandlerContext ctx, HttpRequest request,
+            HttpResponse response, URL requestUri) throws IOException {
+          super.verifyRequest(appid, ctx, request, response, requestUri);
+        }
+
+        @Override
+        protected MapOutputInfo getMapOutputInfo(String dagId, String mapId, Range reduceRange,
+            String jobId, String user) {
+          return null;
+        }
+
+        @Override
+        protected void populateHeaders(List<String> mapIds, String jobId, String dagId, String user,
+            Range reduceRange, HttpResponse response, boolean keepAliveParam,
+            Map<String, MapOutputInfo> infoMap) throws IOException {
+          throw new DiskErrorException(MESSAGE);
+        }
+
+        @Override
+        protected ChannelFuture sendMapOutput(ChannelHandlerContext ctx, Channel ch, String user,
+            String mapId, Range reduceRange, MapOutputInfo info) throws IOException {
+          return null;
+        }
+      };
+    }
+
+    public String getSecret() {
+      return "secret";
+    }
+  }
+
   /**
    * Test the validation of ShuffleHandler's meta-data's serialization and
    * de-serialization.
@@ -1322,6 +1369,53 @@
     sh.close();
   }
 
+  @Test
+  public void testShuffleHandlerSendsDiskError() throws Exception {
+    Configuration conf = new Configuration();
+    conf.setInt(ShuffleHandler.SHUFFLE_PORT_CONFIG_KEY, 0);
+
+    DataInputStream input = null;
+    MockShuffleHandlerWithFatalDiskError shuffleHandler =
+        new MockShuffleHandlerWithFatalDiskError();
+    try {
+      shuffleHandler.init(conf);
+      shuffleHandler.start();
+
+      String shuffleBaseURL = "http://127.0.0.1:"
+          + shuffleHandler.getConfig().get(ShuffleHandler.SHUFFLE_PORT_CONFIG_KEY);
+      URL url = new URL(
+          shuffleBaseURL + "/mapOutput?job=job_12345_1&dag=1&reduce=1&map=attempt_12345_1_m_1_0");
+      shuffleHandler.secretManager.addTokenForJob("job_12345_1",
+          new Token<>("id".getBytes(), shuffleHandler.getSecret().getBytes(), null, null));
+
+      HttpConnectionParams httpConnectionParams = ShuffleUtils.getHttpConnectionParams(conf);
+      BaseHttpConnection httpConnection = ShuffleUtils.getHttpConnection(true, url,
+          httpConnectionParams, "testFetcher", shuffleHandler.secretManager);
+
+      boolean connectSucceeded = httpConnection.connect();
+      Assert.assertTrue(connectSucceeded);
+
+      input = httpConnection.getInputStream();
+      httpConnection.validate();
+
+      ShuffleHeader header = new ShuffleHeader();
+      header.readFields(input);
+
+      // message is encoded in the shuffle header, and can be checked by fetchers
+      Assert.assertEquals(
+          ShuffleHandlerError.DISK_ERROR_EXCEPTION + ": " + MockShuffleHandlerWithFatalDiskError.MESSAGE,
+          header.getMapId());
+      Assert.assertEquals(-1, header.getCompressedLength());
+      Assert.assertEquals(-1, header.getUncompressedLength());
+      Assert.assertEquals(-1, header.getPartition());
+    } finally {
+      if (input != null) {
+        input.close();
+      }
+      shuffleHandler.close();
+    }
+  }
+
   public ChannelFuture createMockChannelFuture(Channel mockCh,
       final List<ShuffleHandler.ReduceMapFileCount> listenerList) {
     final ChannelFuture mockFuture = mock(ChannelFuture.class);
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezEvent.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezEvent.java
index e7af4a1..ebea9a4 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezEvent.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezEvent.java
@@ -191,6 +191,8 @@
             .setIndex(ideEvt.getIndex())
             .setDiagnostics(ideEvt.getDiagnostics())
             .setVersion(ideEvt.getVersion())
+            .setIsLocalFetch(ideEvt.isLocalFetch())
+            .setIsDiskErrorAtSource(ideEvt.isDiskErrorAtSource())
             .build();
         break;
       case TASK_ATTEMPT_FAILED_EVENT:
@@ -294,10 +296,9 @@
         event = ProtoConverters.convertVertexManagerEventFromProto(vmProto);
         break;
       case INPUT_READ_ERROR_EVENT:
-        InputReadErrorEventProto ideProto =
-            InputReadErrorEventProto.parseFrom(input);
-        event = InputReadErrorEvent.create(ideProto.getDiagnostics(),
-            ideProto.getIndex(), ideProto.getVersion());
+        InputReadErrorEventProto ideProto = InputReadErrorEventProto.parseFrom(input);
+        event = InputReadErrorEvent.create(ideProto.getDiagnostics(), ideProto.getIndex(),
+            ideProto.getVersion(), ideProto.getIsLocalFetch(), ideProto.getIsDiskErrorAtSource());
         break;
       case TASK_ATTEMPT_FAILED_EVENT:
         TaskAttemptFailedEventProto tfProto =
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
index 18f66cc..f295c06 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
@@ -66,7 +66,7 @@
 import org.apache.tez.runtime.library.common.sort.impl.TezSpillRecord;
 import org.apache.tez.runtime.library.exceptions.FetcherReadTimeoutException;
 import org.apache.tez.runtime.library.common.shuffle.FetchedInput.Type;
-
+import org.apache.tez.runtime.library.common.shuffle.api.ShuffleHandlerError;
 import org.apache.tez.common.Preconditions;
 
 /**
@@ -277,7 +277,8 @@
 
     HostFetchResult hostFetchResult;
 
-    if (localDiskFetchEnabled && host.equals(localHostname) && port == shufflePort) {
+    boolean isLocalFetch = localDiskFetchEnabled && host.equals(localHostname) && port == shufflePort;
+    if (isLocalFetch) {
       hostFetchResult = setupLocalDiskFetch();
     } else if (multiplex) {
       hostFetchResult = doSharedFetch();
@@ -288,7 +289,7 @@
     if (hostFetchResult.failedInputs != null && hostFetchResult.failedInputs.length > 0) {
       if (!isShutDown.get()) {
         LOG.warn("copyInputs failed for tasks " + Arrays.toString(hostFetchResult.failedInputs));
-        for (InputAttemptIdentifier left : hostFetchResult.failedInputs) {
+        for (InputAttemptFetchFailure left : hostFetchResult.failedInputs) {
           fetcherCallback.fetchFailed(host, left, hostFetchResult.connectFailed);
         }
       } else {
@@ -504,7 +505,7 @@
       // ioErrs.increment(1);
       // If connect did not succeed, just mark all the maps as failed,
       // indirectly penalizing the host
-      InputAttemptIdentifier[] failedFetches = null;
+      InputAttemptFetchFailure[] failedFetches = null;
       if (isShutDown.get()) {
         if (isDebugEnabled) {
           LOG.debug(
@@ -512,8 +513,7 @@
                   e.getClass().getName() + ", Message: " + e.getMessage());
         }
       } else {
-        failedFetches = srcAttemptsRemaining.values().
-            toArray(new InputAttemptIdentifier[srcAttemptsRemaining.values().size()]);
+        failedFetches = InputAttemptFetchFailure.fromAttempts(srcAttemptsRemaining.values());
       }
       return new HostFetchResult(new FetchResult(host, port, partition, partitionCount, srcAttemptsRemaining.values()), failedFetches, true);
     }
@@ -547,7 +547,7 @@
             "Fetch Failure while connecting from %s to: %s:%d, attempt: %s Informing ShuffleManager: ",
             localHostname, host, port, firstAttempt), e);
         return new HostFetchResult(new FetchResult(host, port, partition, partitionCount, srcAttemptsRemaining.values()),
-            new InputAttemptIdentifier[] { firstAttempt }, true);
+            new InputAttemptFetchFailure[] { new InputAttemptFetchFailure(firstAttempt) }, true);
       }
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt(); //reset status
@@ -584,7 +584,7 @@
     // On any error, faildTasks is not null and we exit
     // after putting back the remaining maps to the
     // yet_to_be_fetched list and marking the failed tasks.
-    InputAttemptIdentifier[] failedInputs = null;
+    InputAttemptFetchFailure[] failedInputs = null;
     while (!srcAttemptsRemaining.isEmpty() && failedInputs == null) {
       InputAttemptIdentifier inputAttemptIdentifier =
           srcAttemptsRemaining.entrySet().iterator().next().getValue();
@@ -711,7 +711,7 @@
       }
     }
 
-    InputAttemptIdentifier[] failedFetches = null;
+    InputAttemptFetchFailure[] failedFetches = null;
     if (failMissing && srcAttemptsRemaining.size() > 0) {
       if (isShutDown.get()) {
         if (isDebugEnabled) {
@@ -720,8 +720,8 @@
                   " remaining inputs");
         }
       } else {
-        failedFetches = srcAttemptsRemaining.values().
-            toArray(new InputAttemptIdentifier[srcAttemptsRemaining.values().size()]);
+        failedFetches =
+            InputAttemptFetchFailure.fromAttemptsLocalFetchFailure(srcAttemptsRemaining.values());
       }
     } else {
       // nothing needs to be done to requeue remaining entries
@@ -770,10 +770,10 @@
 
   static class HostFetchResult {
     private final FetchResult fetchResult;
-    private final InputAttemptIdentifier[] failedInputs;
+    private final InputAttemptFetchFailure[] failedInputs;
     private final boolean connectFailed;
 
-    public HostFetchResult(FetchResult fetchResult, InputAttemptIdentifier[] failedInputs,
+    public HostFetchResult(FetchResult fetchResult, InputAttemptFetchFailure[] failedInputs,
                            boolean connectFailed) {
       this.fetchResult = fetchResult;
       this.failedInputs = failedInputs;
@@ -831,8 +831,11 @@
       return "id: " + srcAttemptId + ", decompressed length: " + decompressedLength + ", compressed length: " + compressedLength + ", reduce: " + forReduce;
     }
   }
-  private InputAttemptIdentifier[] fetchInputs(DataInputStream input,
-      CachingCallBack callback, InputAttemptIdentifier inputAttemptIdentifier) throws FetcherReadTimeoutException {
+
+  @VisibleForTesting
+  InputAttemptFetchFailure[] fetchInputs(DataInputStream input, CachingCallBack callback,
+      InputAttemptIdentifier inputAttemptIdentifier)
+      throws FetcherReadTimeoutException {
     FetchedInput fetchedInput = null;
     InputAttemptIdentifier srcAttemptId = null;
     long decompressedLength = 0;
@@ -856,9 +859,19 @@
           header.readFields(input);
           pathComponent = header.getMapId();
           if (!pathComponent.startsWith(InputAttemptIdentifier.PATH_PREFIX)) {
-            throw new IllegalArgumentException("Invalid map id: " + header.getMapId() + ", expected to start with " +
-                InputAttemptIdentifier.PATH_PREFIX + ", partition: " + header.getPartition()
-                + " while fetching " + inputAttemptIdentifier);
+            if (pathComponent.startsWith(ShuffleHandlerError.DISK_ERROR_EXCEPTION.toString())) {
+              LOG.warn("Invalid map id: " + header.getMapId() + ", expected to start with "
+                  + InputAttemptIdentifier.PATH_PREFIX + ", partition: " + header.getPartition()
+                  + " while fetching " + inputAttemptIdentifier);
+              // this should be treated as local fetch failure while reporting later
+              return new InputAttemptFetchFailure[] {
+                  InputAttemptFetchFailure.fromDiskErrorAtSource(inputAttemptIdentifier) };
+            } else {
+              throw new IllegalArgumentException(
+                  "Invalid map id: " + header.getMapId() + ", expected to start with "
+                      + InputAttemptIdentifier.PATH_PREFIX + ", partition: " + header.getPartition()
+                      + " while fetching " + inputAttemptIdentifier);
+            }
           }
 
           srcAttemptId = pathToAttemptMap.get(new PathPartition(pathComponent, header.getPartition()));
@@ -883,7 +896,7 @@
           if (!isShutDown.get()) {
             LOG.warn("Invalid src id ", e);
             // Don't know which one was bad, so consider all of them as bad
-            return srcAttemptsRemaining.values().toArray(new InputAttemptIdentifier[srcAttemptsRemaining.size()]);
+            return InputAttemptFetchFailure.fromAttempts(srcAttemptsRemaining.values());
           } else {
             if (isDebugEnabled) {
               LOG.debug("Already shutdown. Ignoring badId error with message: " + e.getMessage());
@@ -902,7 +915,8 @@
               srcAttemptId = getNextRemainingAttempt();
             }
             assert (srcAttemptId != null);
-            return new InputAttemptIdentifier[]{srcAttemptId};
+            return new InputAttemptFetchFailure[] {
+                InputAttemptFetchFailure.fromAttempt(srcAttemptId) };
           } else {
             if (isDebugEnabled) {
               LOG.debug("Already shutdown. Ignoring verification failure.");
@@ -1004,10 +1018,10 @@
         // Cleanup the fetchedInput before returning.
         cleanupFetchedInput(fetchedInput);
         if (srcAttemptId == null) {
-          return srcAttemptsRemaining.values()
-              .toArray(new InputAttemptIdentifier[srcAttemptsRemaining.size()]);
+          return InputAttemptFetchFailure.fromAttempts(srcAttemptsRemaining.values());
         } else {
-          return new InputAttemptIdentifier[] { srcAttemptId };
+          return new InputAttemptFetchFailure[] {
+              new InputAttemptFetchFailure(srcAttemptId) };
         }
       }
       LOG.warn("Failed to shuffle output of " + srcAttemptId + " from " + host + " (to "
@@ -1016,7 +1030,8 @@
       // Cleanup the fetchedInput
       cleanupFetchedInput(fetchedInput);
       // metrics.failedFetch();
-      return new InputAttemptIdentifier[] { srcAttemptId };
+      return new InputAttemptFetchFailure[] {
+          new InputAttemptFetchFailure(srcAttemptId) };
     }
     return null;
   }
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/FetcherCallback.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/FetcherCallback.java
index 34bd272..b751fb9 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/FetcherCallback.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/FetcherCallback.java
@@ -28,6 +28,7 @@
       FetchedInput fetchedInput, long fetchedBytes, long decompressedLength, long copyDuration)
       throws IOException;
   
-  public void fetchFailed(String host, InputAttemptIdentifier srcAttemptIdentifier, boolean connectFailed);
+  public void fetchFailed(String host, InputAttemptFetchFailure srcAttemptFetchFailure,
+      boolean connectFailed);
 
 }
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/InputAttemptFetchFailure.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/InputAttemptFetchFailure.java
new file mode 100644
index 0000000..d94db35
--- /dev/null
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/InputAttemptFetchFailure.java
@@ -0,0 +1,115 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.tez.runtime.library.common.shuffle;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.apache.tez.runtime.library.common.CompositeInputAttemptIdentifier;
+import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+
+/**
+ * InputAttemptFetchFailure is supposed to wrap an InputAttemptIdentifier with any kind of failure
+ * information during fetch. It can be useful for propagating as a single object instead of multiple
+ * parameters (local fetch error, remote fetch error, connect failed, read failed, etc.).
+ */
+public class InputAttemptFetchFailure {
+
+  private final InputAttemptIdentifier inputAttemptIdentifier;
+  private final boolean isLocalFetch;
+  private final boolean isDiskErrorAtSource;
+
+  public InputAttemptFetchFailure(InputAttemptIdentifier inputAttemptIdentifier) {
+    this(inputAttemptIdentifier, false, false);
+  }
+
+  public InputAttemptFetchFailure(InputAttemptIdentifier inputAttemptIdentifier,
+      boolean isLocalFetch, boolean isDiskErrorAtSource) {
+    this.inputAttemptIdentifier = inputAttemptIdentifier;
+    this.isLocalFetch = isLocalFetch;
+    this.isDiskErrorAtSource = isDiskErrorAtSource;
+  }
+
+  public InputAttemptIdentifier getInputAttemptIdentifier() {
+    return inputAttemptIdentifier;
+  }
+
+  public boolean isLocalFetch() {
+    return isLocalFetch;
+  }
+
+  public boolean isDiskErrorAtSource() {
+    return isDiskErrorAtSource;
+  }
+
+  public static InputAttemptFetchFailure fromAttempt(InputAttemptIdentifier attempt) {
+    return new InputAttemptFetchFailure(attempt, false, false);
+  }
+
+  public static InputAttemptFetchFailure fromLocalFetchFailure(InputAttemptIdentifier attempt) {
+    return new InputAttemptFetchFailure(attempt, true, false);
+  }
+
+  public static InputAttemptFetchFailure fromDiskErrorAtSource(InputAttemptIdentifier attempt) {
+    return new InputAttemptFetchFailure(attempt, false, true);
+  }
+
+  public static InputAttemptFetchFailure[] fromAttempts(Collection<InputAttemptIdentifier> values) {
+    return values.stream().map(identifier -> new InputAttemptFetchFailure(identifier, false, false))
+        .toArray(InputAttemptFetchFailure[]::new);
+  }
+
+  public static InputAttemptFetchFailure[] fromAttempts(InputAttemptIdentifier[] values) {
+    return Arrays.asList(values).stream()
+        .map(identifier -> new InputAttemptFetchFailure(identifier, false, false))
+        .toArray(InputAttemptFetchFailure[]::new);
+  }
+
+  public static InputAttemptFetchFailure[] fromAttemptsLocalFetchFailure(
+      Collection<InputAttemptIdentifier> values) {
+    return values.stream().map(identifier -> new InputAttemptFetchFailure(identifier, true, false))
+        .toArray(InputAttemptFetchFailure[]::new);
+  }
+
+  public static InputAttemptFetchFailure fromCompositeAttemptLocalFetchFailure(
+      CompositeInputAttemptIdentifier compositeInputAttemptIdentifier) {
+    return new InputAttemptFetchFailure(compositeInputAttemptIdentifier, true, false);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (obj == null || (obj.getClass() != this.getClass())) {
+      return false;
+    }
+    return inputAttemptIdentifier.equals(((InputAttemptFetchFailure) obj).inputAttemptIdentifier)
+        && isLocalFetch == ((InputAttemptFetchFailure) obj).isLocalFetch
+        && isDiskErrorAtSource == ((InputAttemptFetchFailure) obj).isDiskErrorAtSource;
+  }
+
+  @Override
+  public int hashCode() {
+    return 31 * inputAttemptIdentifier.hashCode() + 31 * (isLocalFetch ? 0 : 1)
+        + 31 * (isDiskErrorAtSource ? 0 : 1);
+  }
+
+  @Override
+  public String toString() {
+    return String.format("%s, isLocalFetch: %s, isDiskErrorAtSource: %s",
+        inputAttemptIdentifier.toString(), isLocalFetch, isDiskErrorAtSource);
+  }
+}
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
index 742fc18..2b83ad8 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
@@ -84,6 +84,7 @@
 import org.apache.tez.runtime.library.common.shuffle.Fetcher.FetcherBuilder;
 import org.apache.tez.runtime.library.common.shuffle.FetcherCallback;
 import org.apache.tez.runtime.library.common.shuffle.HostPort;
+import org.apache.tez.runtime.library.common.shuffle.InputAttemptFetchFailure;
 import org.apache.tez.runtime.library.common.shuffle.InputHost;
 import org.apache.tez.runtime.library.common.shuffle.InputHost.PartitionToInputs;
 import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
@@ -391,9 +392,9 @@
               List<Event> failedEventsToSend = Lists.newArrayListWithCapacity(
                   failedEvents.size());
               for (InputReadErrorEvent key : failedEvents.keySet()) {
-                failedEventsToSend.add(InputReadErrorEvent
-                    .create(key.getDiagnostics(), key.getIndex(),
-                        key.getVersion(), failedEvents.get(key)));
+                failedEventsToSend.add(InputReadErrorEvent.create(key.getDiagnostics(),
+                    key.getIndex(), key.getVersion(), failedEvents.get(key), key.isLocalFetch(),
+                    key.isDiskErrorAtSource()));
               }
               inputContext.sendEvents(failedEventsToSend);
               failedEvents.clear();
@@ -939,12 +940,15 @@
 
   @Override
   public void fetchFailed(String host,
-      InputAttemptIdentifier srcAttemptIdentifier, boolean connectFailed) {
+      InputAttemptFetchFailure inputAttemptFetchFailure, boolean connectFailed) {
     // TODO NEWTEZ. Implement logic to report fetch failures after a threshold.
     // For now, reporting immediately.
-    LOG.info(srcNameTrimmed + ": " + "Fetch failed for src: " + srcAttemptIdentifier
-        + "InputIdentifier: " + srcAttemptIdentifier + ", connectFailed: "
-        + connectFailed);
+    InputAttemptIdentifier srcAttemptIdentifier = inputAttemptFetchFailure.getInputAttemptIdentifier();
+    LOG.info(
+        "{}: Fetch failed for src: {} InputIdentifier: {}, connectFailed: {}, "
+            + "local fetch: {}, remote fetch failure reported as local failure: {})",
+        srcNameTrimmed, srcAttemptIdentifier, srcAttemptIdentifier, connectFailed,
+        inputAttemptFetchFailure.isLocalFetch(), inputAttemptFetchFailure.isDiskErrorAtSource());
     failedShufflesCounter.increment(1);
     inputContext.notifyProgress();
     if (srcAttemptIdentifier == null) {
@@ -957,7 +961,9 @@
               srcAttemptIdentifier.getInputIdentifier(),
               srcAttemptIdentifier.getAttemptNumber()),
           srcAttemptIdentifier.getInputIdentifier(),
-          srcAttemptIdentifier.getAttemptNumber());
+          srcAttemptIdentifier.getAttemptNumber(),
+          inputAttemptFetchFailure.isLocalFetch(),
+          inputAttemptFetchFailure.isDiskErrorAtSource());
       if (maxTimeToWaitForReportMillis > 0) {
         try {
           reportLock.lock();
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java
index e732ab0..3272327 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java
@@ -50,7 +50,9 @@
 import org.apache.tez.runtime.library.common.sort.impl.TezIndexRecord;
 import org.apache.tez.runtime.library.common.sort.impl.TezSpillRecord;
 import org.apache.tez.runtime.library.exceptions.FetcherReadTimeoutException;
+import org.apache.tez.runtime.library.common.shuffle.InputAttemptFetchFailure;
 import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
+import org.apache.tez.runtime.library.common.shuffle.api.ShuffleHandlerError;
 
 import com.google.common.annotations.VisibleForTesting;
 
@@ -272,7 +274,8 @@
       // On any error, faildTasks is not null and we exit
       // after putting back the remaining maps to the 
       // yet_to_be_fetched list and marking the failed tasks.
-      InputAttemptIdentifier[] failedTasks = null;
+      InputAttemptFetchFailure[] failedTasks = null;
+
       while (!remaining.isEmpty() && failedTasks == null) {
         InputAttemptIdentifier inputAttemptIdentifier =
             remaining.entrySet().iterator().next().getValue();
@@ -300,25 +303,14 @@
               }
               return;
             }
-            failedTasks = new InputAttemptIdentifier[] {getNextRemainingAttempt()};
+            failedTasks = new InputAttemptFetchFailure[] {
+                new InputAttemptFetchFailure(getNextRemainingAttempt()) };
             break;
           }
         }
       }
 
-      if (failedTasks != null && failedTasks.length > 0) {
-        if (stopped) {
-          if (LOG.isDebugEnabled()) {
-            LOG.debug("Ignoring copyMapOutput failures for tasks: " + Arrays.toString(failedTasks) +
-                " since Fetcher has been stopped");
-          }
-        } else {
-          LOG.warn("copyMapOutput failed for tasks " + Arrays.toString(failedTasks));
-          for (InputAttemptIdentifier left : failedTasks) {
-            scheduler.copyFailed(left, host, true, false, false);
-          }
-        }
-      }
+      invokeCopyFailedForFailedTasks(host, failedTasks);
 
       cleanupCurrentConnection(false);
 
@@ -332,6 +324,23 @@
     }
   }
 
+  private void invokeCopyFailedForFailedTasks(MapHost host,
+      InputAttemptFetchFailure[] failedTasks) {
+    if (failedTasks != null && failedTasks.length > 0) {
+      if (stopped) {
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Ignoring copyMapOutput failures for tasks: " + Arrays.toString(failedTasks)
+              + " since Fetcher has been stopped");
+        }
+      } else {
+        LOG.warn("copyMapOutput failed for tasks " + Arrays.toString(failedTasks));
+        for (InputAttemptFetchFailure left : failedTasks) {
+          scheduler.copyFailed(left, host, true, false);
+        }
+      }
+    }
+  }
+
   @VisibleForTesting
   boolean setupConnection(MapHost host, Collection<InputAttemptIdentifier> attempts)
       throws IOException {
@@ -380,7 +389,8 @@
       for (InputAttemptIdentifier left : remaining.values()) {
         // Need to be handling temporary glitches ..
         // Report read error to the AM to trigger source failure heuristics
-        scheduler.copyFailed(left, host, connectSucceeded, !connectSucceeded, false);
+        scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(left), host, connectSucceeded,
+            !connectSucceeded);
       }
       return false;
     }
@@ -404,7 +414,8 @@
     }
   }
 
-  private static InputAttemptIdentifier[] EMPTY_ATTEMPT_ID_ARRAY = new InputAttemptIdentifier[0];
+  private static final InputAttemptFetchFailure[] EMPTY_ATTEMPT_ID_ARRAY =
+      new InputAttemptFetchFailure[0];
 
   private static class MapOutputStat {
     final InputAttemptIdentifier srcAttemptId;
@@ -425,8 +436,8 @@
     }
   }
 
-  protected InputAttemptIdentifier[] copyMapOutput(MapHost host,
-                                DataInputStream input, InputAttemptIdentifier inputAttemptIdentifier) throws FetcherReadTimeoutException {
+  protected InputAttemptFetchFailure[] copyMapOutput(MapHost host, DataInputStream input,
+      InputAttemptIdentifier inputAttemptIdentifier) throws FetcherReadTimeoutException {
     MapOutput mapOutput = null;
     InputAttemptIdentifier srcAttemptId = null;
     long decompressedLength = 0;
@@ -452,7 +463,13 @@
               badIdErrs.increment(1);
               LOG.warn("Invalid map id: " + header.mapId + ", expected to start with " +
                   InputAttemptIdentifier.PATH_PREFIX + ", partition: " + header.forReduce);
-              return new InputAttemptIdentifier[]{getNextRemainingAttempt()};
+              if (header.mapId.startsWith(ShuffleHandlerError.DISK_ERROR_EXCEPTION.toString())) {
+                //this should be treated as local fetch failure while reporting later
+                return new InputAttemptFetchFailure[] {
+                    InputAttemptFetchFailure.fromDiskErrorAtSource(getNextRemainingAttempt()) };
+              }
+              return new InputAttemptFetchFailure[] {
+                  InputAttemptFetchFailure.fromAttempt(getNextRemainingAttempt()) };
             } else {
               if (LOG.isDebugEnabled()) {
                 LOG.debug("Already shutdown. Ignoring invalid map id error");
@@ -477,7 +494,8 @@
             LOG.warn("Invalid map id ", e);
             // Don't know which one was bad, so consider this one bad and dont read
             // the remaining because we dont know where to start reading from. YARN-1773
-            return new InputAttemptIdentifier[]{getNextRemainingAttempt()};
+            return new InputAttemptFetchFailure[] {
+                new InputAttemptFetchFailure(getNextRemainingAttempt()) };
           } else {
             if (LOG.isDebugEnabled()) {
               LOG.debug("Already shutdown. Ignoring invalid map id error. Exception: " +
@@ -497,7 +515,8 @@
               LOG.warn("Was expecting " + srcAttemptId + " but got null");
             }
             assert (srcAttemptId != null);
-            return new InputAttemptIdentifier[]{srcAttemptId};
+            return new InputAttemptFetchFailure[] {
+                new InputAttemptFetchFailure(getNextRemainingAttempt()) };
           } else {
             if (LOG.isDebugEnabled()) {
               LOG.debug("Already stopped. Ignoring verification failure.");
@@ -595,9 +614,10 @@
             srcAttemptId + " decomp: " +
             decompressedLength + ", " + compressedLength, ioe);
         if (srcAttemptId == null) {
-          return remaining.values().toArray(new InputAttemptIdentifier[remaining.values().size()]);
+          return InputAttemptFetchFailure.fromAttempts(remaining.values());
         } else {
-          return new InputAttemptIdentifier[]{srcAttemptId};
+          return new InputAttemptFetchFailure[] {
+              new InputAttemptFetchFailure(srcAttemptId) };
         }
       }
       LOG.warn("Failed to shuffle output of " + srcAttemptId +
@@ -605,7 +625,8 @@
 
       // Inform the shuffle-scheduler
       mapOutput.abort();
-      return new InputAttemptIdentifier[] {srcAttemptId};
+      return new InputAttemptFetchFailure[] {
+          new InputAttemptFetchFailure(srcAttemptId) };
     }
     return null;
   }
@@ -734,7 +755,8 @@
             if (!stopped) {
               hasFailures = true;
               ioErrs.increment(1);
-              scheduler.copyFailed(srcAttemptId, host, true, false, true);
+              scheduler.copyFailed(InputAttemptFetchFailure.fromLocalFetchFailure(srcAttemptId),
+                  host, true, false);
               LOG.warn("Failed to read local disk output of " + srcAttemptId + " from " +
                   host.getHostIdentifier(), e);
             } else {
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleHeader.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleHeader.java
index 9f883db..f074e89 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleHeader.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleHeader.java
@@ -103,4 +103,11 @@
     WritableUtils.writeVLong(out, uncompressedLength);
     WritableUtils.writeVInt(out, forReduce);
   }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "ShuffleHeader [mapId=%s, uncompressedLength=%d, compressedLength=%d, forReduce=%d]", mapId,
+        uncompressedLength, compressedLength, forReduce);
+  }
 }
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java
index 0954a76..e7f63ab 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java
@@ -80,6 +80,7 @@
 import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
 import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils.FetchStatsLogger;
 import org.apache.tez.runtime.library.common.shuffle.HostPort;
+import org.apache.tez.runtime.library.common.shuffle.InputAttemptFetchFailure;
 import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.MapHost.HostPortPartition;
 import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.MapOutput.Type;
 
@@ -755,16 +756,13 @@
     }
   }
 
-  public synchronized void copyFailed(InputAttemptIdentifier srcAttempt,
-                                      MapHost host,
-                                      boolean readError,
-                                      boolean connectError,
-                                      boolean isLocalFetch) {
+  public synchronized void copyFailed(InputAttemptFetchFailure fetchFailure, MapHost host,
+      boolean readError, boolean connectError) {
     failedShuffleCounter.increment(1);
     inputContext.notifyProgress();
-    int failures = incrementAndGetFailureAttempt(srcAttempt);
+    int failures = incrementAndGetFailureAttempt(fetchFailure.getInputAttemptIdentifier());
 
-    if (!isLocalFetch) {
+    if (!fetchFailure.isLocalFetch()) {
       /**
        * Track the number of failures that has happened since last completion.
        * This gets reset on a successful copy.
@@ -789,11 +787,11 @@
 
     if (shouldInformAM) {
       //Inform AM. In case producer needs to be restarted, it is handled at AM.
-      informAM(srcAttempt);
+      informAM(fetchFailure);
     }
 
     //Restart consumer in case shuffle is not healthy
-    if (!isShuffleHealthy(srcAttempt)) {
+    if (!isShuffleHealthy(fetchFailure.getInputAttemptIdentifier())) {
       return;
     }
 
@@ -868,21 +866,24 @@
   }
 
   // Notify AM
-  private void informAM(InputAttemptIdentifier srcAttempt) {
+  private void informAM(InputAttemptFetchFailure fetchFailure) {
+    InputAttemptIdentifier srcAttempt = fetchFailure.getInputAttemptIdentifier();
     LOG.info(
-        srcNameTrimmed + ": " + "Reporting fetch failure for InputIdentifier: "
-            + srcAttempt + " taskAttemptIdentifier: " + TezRuntimeUtils
-            .getTaskAttemptIdentifier(inputContext.getSourceVertexName(),
-                srcAttempt.getInputIdentifier(),
-                srcAttempt.getAttemptNumber()) + " to AM.");
+        "{}: Reporting fetch failure for InputIdentifier: {} taskAttemptIdentifier: {}, "
+            + "local fetch: {}, remote fetch failure reported as local failure: {}) to AM.",
+        srcNameTrimmed, srcAttempt,
+        TezRuntimeUtils.getTaskAttemptIdentifier(inputContext.getSourceVertexName(),
+            srcAttempt.getInputIdentifier(), srcAttempt.getAttemptNumber()),
+        fetchFailure.isLocalFetch(), fetchFailure.isDiskErrorAtSource());
     List<Event> failedEvents = Lists.newArrayListWithCapacity(1);
-    failedEvents.add(InputReadErrorEvent.create(
-        "Fetch failure for " + TezRuntimeUtils
-            .getTaskAttemptIdentifier(inputContext.getSourceVertexName(),
-                srcAttempt.getInputIdentifier(),
-                srcAttempt.getAttemptNumber()) + " to jobtracker.",
-        srcAttempt.getInputIdentifier(),
-        srcAttempt.getAttemptNumber()));
+    failedEvents.add(
+        InputReadErrorEvent.create(
+            "Fetch failure for "
+                + TezRuntimeUtils.getTaskAttemptIdentifier(inputContext.getSourceVertexName(),
+                    srcAttempt.getInputIdentifier(), srcAttempt.getAttemptNumber())
+                + " to jobtracker.",
+            srcAttempt.getInputIdentifier(), srcAttempt.getAttemptNumber(),
+            fetchFailure.isLocalFetch(), fetchFailure.isDiskErrorAtSource()));
 
     inputContext.sendEvents(failedEvents);
   }
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java
index db9c7af..05d4eb4 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/TestFetcher.java
@@ -32,6 +32,7 @@
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 
+import java.io.DataInputStream;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -46,10 +47,16 @@
 import org.apache.hadoop.fs.LocalFileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.tez.common.counters.TezCounters;
 import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.runtime.api.InputContext;
 import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
 import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.apache.tez.runtime.library.common.shuffle.api.ShuffleHandlerError;
+import org.apache.tez.runtime.library.common.shuffle.impl.ShuffleManager;
+import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.ShuffleHeader;
 import org.apache.tez.runtime.library.common.sort.impl.TezIndexRecord;
+import org.apache.tez.runtime.library.testutils.RuntimeTestUtils;
 import org.junit.Assert;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
@@ -79,7 +86,8 @@
     Fetcher fetcher = spy(builder.build());
 
     FetchResult fr = new FetchResult(HOST, PORT, 0, 1, Arrays.asList(srcAttempts));
-    Fetcher.HostFetchResult hfr = new Fetcher.HostFetchResult(fr, srcAttempts, false);
+    Fetcher.HostFetchResult hfr =
+        new Fetcher.HostFetchResult(fr, InputAttemptFetchFailure.fromAttempts(srcAttempts), false);
     doReturn(hfr).when(fetcher).setupLocalDiskFetch();
     doReturn(null).when(fetcher).doHttpFetch();
     doNothing().when(fetcher).shutdown();
@@ -151,7 +159,7 @@
     };
     final int FIRST_FAILED_ATTEMPT_IDX = 2;
     final int SECOND_FAILED_ATTEMPT_IDX = 4;
-    final int[] sucessfulAttempts = {0, 1, 3};
+    final int[] successfulAttempts = {0, 1, 3};
 
     TezConfiguration conf = new TezConfiguration();
     conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH, "true");
@@ -206,18 +214,24 @@
     doNothing().when(fetcher).shutdown();
     doNothing().when(callback).fetchSucceeded(anyString(), any(InputAttemptIdentifier.class),
         any(FetchedInput.class), anyLong(), anyLong(), anyLong());
-    doNothing().when(callback).fetchFailed(anyString(), any(InputAttemptIdentifier.class), eq(false));
+    doNothing().when(callback).fetchFailed(anyString(), any(InputAttemptFetchFailure.class), eq(false));
 
     FetchResult fetchResult = fetcher.call();
 
     verify(fetcher).setupLocalDiskFetch();
 
-    // expect 3 sucesses and 2 failures
-    for (int i : sucessfulAttempts) {
+    // expect 3 successes and 2 failures
+    for (int i : successfulAttempts) {
       verifyFetchSucceeded(callback, srcAttempts[i], conf);
     }
-    verify(callback).fetchFailed(eq(HOST), eq(srcAttempts[FIRST_FAILED_ATTEMPT_IDX]), eq(false));
-    verify(callback).fetchFailed(eq(HOST), eq(srcAttempts[SECOND_FAILED_ATTEMPT_IDX]), eq(false));
+    verify(callback).fetchFailed(eq(HOST),
+        eq(InputAttemptFetchFailure
+            .fromCompositeAttemptLocalFetchFailure(srcAttempts[FIRST_FAILED_ATTEMPT_IDX])),
+        eq(false));
+    verify(callback).fetchFailed(eq(HOST),
+        eq(InputAttemptFetchFailure
+            .fromCompositeAttemptLocalFetchFailure(srcAttempts[SECOND_FAILED_ATTEMPT_IDX])),
+        eq(false));
 
     Assert.assertEquals("fetchResult host", fetchResult.getHost(), HOST);
     Assert.assertEquals("fetchResult partition", fetchResult.getPartition(), partition);
@@ -304,4 +318,30 @@
       Assert.assertTrue(expectedSrcAttempts[count++].toString().compareTo(key) == 0);
     }
   }
+
+  @Test
+  public void testShuffleHandlerDiskErrorUnordered()
+      throws Exception {
+    Configuration conf = new Configuration();
+
+    InputContext inputContext = mock(InputContext.class);
+    doReturn(new TezCounters()).when(inputContext).getCounters();
+    doReturn("vertex").when(inputContext).getSourceVertexName();
+
+    Fetcher.FetcherBuilder builder = new Fetcher.FetcherBuilder(mock(ShuffleManager.class), null,
+        null, ApplicationId.newInstance(0, 1), 1, null, "fetcherTest", conf, true, HOST, PORT,
+        false, true, false);
+    builder.assignWork(HOST, PORT, 0, 1, Arrays.asList(new InputAttemptIdentifier(0, 0)));
+
+    Fetcher fetcher = builder.build();
+    ShuffleHeader header =
+        new ShuffleHeader(ShuffleHandlerError.DISK_ERROR_EXCEPTION.toString(), -1, -1, -1);
+    DataInputStream input = RuntimeTestUtils.shuffleHeaderToDataInput(header);
+
+    InputAttemptFetchFailure[] failures =
+        fetcher.fetchInputs(input, null, new InputAttemptIdentifier(0, 0));
+    Assert.assertEquals(1, failures.length);
+    Assert.assertTrue(failures[0].isDiskErrorAtSource());
+    Assert.assertFalse(failures[0].isLocalFetch());
+  }
 }
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleManager.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleManager.java
index 94f7f5a..041fd03 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleManager.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestShuffleManager.java
@@ -22,7 +22,6 @@
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyInt;
-import static org.mockito.Matchers.anyList;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.atLeast;
 import static org.mockito.Mockito.doAnswer;
@@ -37,7 +36,6 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
@@ -55,7 +53,6 @@
 import org.apache.tez.common.security.JobTokenIdentifier;
 import org.apache.tez.common.security.JobTokenSecretManager;
 import org.apache.tez.dag.api.TezConfiguration;
-import org.apache.tez.dag.api.TezConstants;
 import org.apache.tez.runtime.api.Event;
 import org.apache.tez.runtime.api.ExecutionContext;
 import org.apache.tez.runtime.api.InputContext;
@@ -66,9 +63,9 @@
 import org.apache.tez.runtime.library.common.shuffle.FetchedInput;
 import org.apache.tez.runtime.library.common.shuffle.FetchedInputAllocator;
 import org.apache.tez.runtime.library.common.shuffle.Fetcher;
+import org.apache.tez.runtime.library.common.shuffle.InputAttemptFetchFailure;
 import org.apache.tez.runtime.library.common.shuffle.FetchResult;
 import org.apache.tez.runtime.library.common.shuffle.InputHost;
-import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
 import org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads.DataMovementEventPayloadProto;
 import org.junit.After;
 import org.junit.Assert;
@@ -234,12 +231,12 @@
         }
       }
     });
-    InputAttemptIdentifier inputAttemptIdentifier
-        = new InputAttemptIdentifier(1, 1);
+    InputAttemptFetchFailure inputAttemptFetchFailure =
+        new InputAttemptFetchFailure(new InputAttemptIdentifier(1, 1));
 
     schedulerGetHostThread.start();
     Thread.sleep(1000);
-    shuffleManager.fetchFailed("host1", inputAttemptIdentifier, false);
+    shuffleManager.fetchFailed("host1", inputAttemptFetchFailure, false);
     Thread.sleep(1000);
 
     ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
@@ -254,8 +251,8 @@
     Assert.assertEquals("Number of failures was: " + inputEvent.getNumFailures(),
         inputEvent.getNumFailures(), 1);
 
-    shuffleManager.fetchFailed("host1", inputAttemptIdentifier, false);
-    shuffleManager.fetchFailed("host1", inputAttemptIdentifier, false);
+    shuffleManager.fetchFailed("host1", inputAttemptFetchFailure, false);
+    shuffleManager.fetchFailed("host1", inputAttemptFetchFailure, false);
 
     Thread.sleep(1000);
     verify(inputContext, times(1)).sendEvents(any());
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestFetcher.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestFetcher.java
index 5f7fe4b..028fbce 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestFetcher.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestFetcher.java
@@ -42,6 +42,7 @@
 import java.net.URL;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
@@ -72,7 +73,10 @@
 import org.apache.tez.runtime.library.common.security.SecureShuffleUtils;
 import org.apache.tez.runtime.library.common.sort.impl.TezIndexRecord;
 import org.apache.tez.runtime.library.exceptions.FetcherReadTimeoutException;
+import org.apache.tez.runtime.library.testutils.RuntimeTestUtils;
+import org.apache.tez.runtime.library.common.shuffle.InputAttemptFetchFailure;
 import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
+import org.apache.tez.runtime.library.common.shuffle.api.ShuffleHandlerError;
 import org.junit.Assert;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
@@ -241,7 +245,7 @@
     );
     final int FIRST_FAILED_ATTEMPT_IDX = 2;
     final int SECOND_FAILED_ATTEMPT_IDX = 4;
-    final int[] sucessfulAttemptsIndexes = { 0, 1, 3 };
+    final int[] successfulAttemptsIndexes = { 0, 1, 3 };
 
     doReturn(srcAttempts).when(scheduler).getMapsForHost(host);
 
@@ -311,13 +315,17 @@
     spyFetcher.setupLocalDiskFetch(host);
 
     // should have exactly 3 success and 1 failure.
-    for (int i : sucessfulAttemptsIndexes) {
+    for (int i : successfulAttemptsIndexes) {
       for (int j = 0; j < host.getPartitionCount(); j++) {
         verifyCopySucceeded(scheduler, host, srcAttempts, i, j);
       }
     }
-    verify(scheduler).copyFailed(srcAttempts.get(FIRST_FAILED_ATTEMPT_IDX).expand(0), host, true, false, true);
-    verify(scheduler).copyFailed(srcAttempts.get(SECOND_FAILED_ATTEMPT_IDX).expand(0), host, true, false, true);
+    verify(scheduler).copyFailed(
+        eq(InputAttemptFetchFailure.fromLocalFetchFailure(srcAttempts.get(FIRST_FAILED_ATTEMPT_IDX).expand(0))),
+        eq(host), eq(true), eq(false));
+    verify(scheduler).copyFailed(
+        eq(InputAttemptFetchFailure.fromLocalFetchFailure(srcAttempts.get(SECOND_FAILED_ATTEMPT_IDX).expand(0))),
+        eq(host), eq(true), eq(false));
 
     verify(spyFetcher).putBackRemainingMapOutputs(host);
     verify(scheduler).putBackKnownMapOutput(host, srcAttempts.get(FIRST_FAILED_ATTEMPT_IDX));
@@ -426,7 +434,7 @@
     );
     final int FIRST_FAILED_ATTEMPT_IDX = 2;
     final int SECOND_FAILED_ATTEMPT_IDX = 4;
-    final int[] sucessfulAttemptsIndexes = { 0, 1, 3 };
+    final int[] successfulAttemptsIndexes = { 0, 1, 3 };
 
     doReturn(srcAttempts).when(scheduler).getMapsForHost(host);
     final ConcurrentMap<ShuffleScheduler.PathPartition, InputAttemptIdentifier> pathToIdentifierMap
@@ -503,15 +511,23 @@
     spyFetcher.setupLocalDiskFetch(host);
 
     // should have exactly 3 success and 1 failure.
-    for (int i : sucessfulAttemptsIndexes) {
+    for (int i : successfulAttemptsIndexes) {
       for (int j = 0; j < host.getPartitionCount(); j++) {
         verifyCopySucceeded(scheduler, host, srcAttempts, i, j);
       }
     }
-    verify(scheduler).copyFailed(srcAttempts.get(FIRST_FAILED_ATTEMPT_IDX).expand(0), host, true, false, true);
-    verify(scheduler).copyFailed(srcAttempts.get(FIRST_FAILED_ATTEMPT_IDX).expand(1), host, true, false, true);
-    verify(scheduler).copyFailed(srcAttempts.get(SECOND_FAILED_ATTEMPT_IDX).expand(0), host, true, false, true);
-    verify(scheduler).copyFailed(srcAttempts.get(SECOND_FAILED_ATTEMPT_IDX).expand(1), host, true, false, true);
+    verify(scheduler).copyFailed(
+        eq(InputAttemptFetchFailure.fromLocalFetchFailure(srcAttempts.get(FIRST_FAILED_ATTEMPT_IDX).expand(0))),
+        eq(host), eq(true), eq(false));
+    verify(scheduler).copyFailed(
+        eq(InputAttemptFetchFailure.fromLocalFetchFailure(srcAttempts.get(FIRST_FAILED_ATTEMPT_IDX).expand(1))),
+        eq(host), eq(true), eq(false));
+    verify(scheduler).copyFailed(eq(
+        InputAttemptFetchFailure.fromLocalFetchFailure(srcAttempts.get(SECOND_FAILED_ATTEMPT_IDX).expand(0))),
+        eq(host), eq(true), eq(false));
+    verify(scheduler).copyFailed(eq(
+        InputAttemptFetchFailure.fromLocalFetchFailure(srcAttempts.get(SECOND_FAILED_ATTEMPT_IDX).expand(1))),
+        eq(host), eq(true), eq(false));
 
     verify(spyFetcher).putBackRemainingMapOutputs(host);
     verify(scheduler).putBackKnownMapOutput(host, srcAttempts.get(FIRST_FAILED_ATTEMPT_IDX));
@@ -630,8 +646,8 @@
     //setup connection should be called twice (1 for connect and another for retry)
     verify(fetcher, times(2)).setupConnection(any(MapHost.class), any(Collection.class));
     //since copyMapOutput consistently fails, it should call copyFailed once
-    verify(scheduler, times(1)).copyFailed(any(InputAttemptIdentifier.class), any(MapHost.class),
-          anyBoolean(), anyBoolean(), anyBoolean());
+    verify(scheduler, times(1)).copyFailed(any(InputAttemptFetchFailure.class), any(MapHost.class),
+          anyBoolean(), anyBoolean());
 
     verify(fetcher, times(1)).putBackRemainingMapOutputs(any(MapHost.class));
     verify(scheduler, times(3)).putBackKnownMapOutput(any(MapHost.class),
@@ -750,6 +766,32 @@
     }
   }
 
+  @Test
+  public void testShuffleHandlerDiskErrorOrdered()
+      throws Exception {
+    MapHost mapHost = new MapHost(HOST, PORT, 0, 1);
+    InputAttemptIdentifier inputAttemptIdentifier = new InputAttemptIdentifier(0, 0, "attempt");
+
+    FetcherOrderedGrouped fetcher = new FetcherOrderedGrouped(null, null, null, null, null, false,
+        0, null, new TezConfiguration(), null, false, HOST, PORT, "src vertex", mapHost,
+        ioErrsCounter, wrongLengthErrsCounter, badIdErrsCounter, wrongMapErrsCounter,
+        connectionErrsCounter, wrongReduceErrsCounter, APP_ID, DAG_ID, false, false, true, false);
+    fetcher.remaining = new HashMap<String, InputAttemptIdentifier>();
+
+    ShuffleHeader header =
+        new ShuffleHeader(ShuffleHandlerError.DISK_ERROR_EXCEPTION.toString(), -1, -1, -1);
+    DataInputStream input = RuntimeTestUtils.shuffleHeaderToDataInput(header);
+
+    // copyMapOutput is used for remote fetch, this time it returns a fetch failure, which is fatal
+    // and should be treated as a local fetch failure
+    InputAttemptFetchFailure[] failures =
+        fetcher.copyMapOutput(mapHost, input, inputAttemptIdentifier);
+
+    Assert.assertEquals(1, failures.length);
+    Assert.assertTrue(failures[0].isDiskErrorAtSource());
+    Assert.assertFalse(failures[0].isLocalFetch());
+  }
+
   private RawLocalFileSystem getRawFs(Configuration conf) {
     try {
       return (RawLocalFileSystem) FileSystem.getLocal(conf).getRaw();
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestShuffleScheduler.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestShuffleScheduler.java
index fabfa27..b89ffb0 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestShuffleScheduler.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestShuffleScheduler.java
@@ -55,6 +55,7 @@
 import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
 import org.apache.tez.runtime.library.common.CompositeInputAttemptIdentifier;
 import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
+import org.apache.tez.runtime.library.common.shuffle.InputAttemptFetchFailure;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -248,8 +249,8 @@
     for (int i = 100; i < 199; i++) {
       InputAttemptIdentifier inputAttemptIdentifier =
           new InputAttemptIdentifier(i, 0, "attempt_");
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
+      scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier),
+          new MapHost("host" + (i % totalProducerNodes), 10000, i, 1), false, true);
     }
 
 
@@ -257,9 +258,8 @@
         new InputAttemptIdentifier(200, 0, "attempt_");
 
     //Should fail here and report exception as reducer is not healthy
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (200 %
-        totalProducerNodes),
-        10000, 200, 1), false, true, false);
+    scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier),
+        new MapHost("host" + (200 % totalProducerNodes), 10000, 200, 1), false, true);
 
     int minFailurePerHost = conf.getInt(
         TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MIN_FAILURES_PER_HOST,
@@ -330,8 +330,8 @@
     for (int i = 190; i < 200; i++) {
       InputAttemptIdentifier inputAttemptIdentifier =
           new InputAttemptIdentifier(i, 0, "attempt_");
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
+      scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier),
+          new MapHost("host" + (i % totalProducerNodes), 10000, i, 1), false, true);
     }
 
     //Shuffle has not stalled. so no issues.
@@ -342,9 +342,8 @@
 
     InputAttemptIdentifier inputAttemptIdentifier =
         new InputAttemptIdentifier(190, 0, "attempt_");
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" +
-        (190 % totalProducerNodes),
-        10000, 190, 1), false, true, false);
+    scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier),
+        new MapHost("host" + (190 % totalProducerNodes), 10000, 190, 1), false, true);
 
     //Even when it is stalled, need (320 - 300 = 20) * 3 = 60 failures
     verify(scheduler.reporter, times(0)).reportException(any(Throwable.class));
@@ -355,16 +354,17 @@
     for (int i = 190; i < 200; i++) {
       inputAttemptIdentifier =
           new InputAttemptIdentifier(i, 0, "attempt_");
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
+      InputAttemptFetchFailure failure = InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes),
+          10000, i, 1), false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes),
+          10000, i, 1), false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes),
+          10000, i, 1), false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes),
+          10000, i, 1), false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes),
+          10000, i, 1), false, true);
     }
 
     assertEquals(61, scheduler.failedShufflesSinceLastCompletion);
@@ -376,12 +376,14 @@
     for (int i = 110; i < 120; i++) {
       inputAttemptIdentifier =
           new InputAttemptIdentifier(i, 0, "attempt_");
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
+      InputAttemptFetchFailure failure =
+          InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
+          false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
+          false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
+          false, true);
     }
 
     // Should fail now due to fetcherHealthy. (stall has already happened and
@@ -432,8 +434,8 @@
     //1 fails (last fetch)
     InputAttemptIdentifier inputAttemptIdentifier =
         new InputAttemptIdentifier(319, 0, "attempt_");
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (319 % totalProducerNodes),
-        10000, 319, 1), false, true, false);
+    scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier),
+        new MapHost("host" + (319 % totalProducerNodes), 10000, 319, 1), false, true);
 
     //stall the shuffle
     scheduler.lastProgressTime = System.currentTimeMillis() - 1000000;
@@ -441,15 +443,13 @@
     assertEquals(scheduler.remainingMaps.get(), 1);
 
     //Retry for 3 more times
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (319 %
-        totalProducerNodes),
-        10000, 319, 1), false, true, false);
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (319 %
-        totalProducerNodes),
-        10000, 310, 1), false, true, false);
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (319 %
-        totalProducerNodes),
-        10000, 310, 1), false, true, false);
+    InputAttemptFetchFailure failure = InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier);
+    scheduler.copyFailed(failure, new MapHost("host" + (319 % totalProducerNodes), 10000, 319, 1),
+        false, true);
+    scheduler.copyFailed(failure, new MapHost("host" + (319 % totalProducerNodes), 10000, 310, 1),
+        false, true);
+    scheduler.copyFailed(failure, new MapHost("host" + (319 % totalProducerNodes), 10000, 310, 1),
+        false, true);
 
     // failedShufflesSinceLastCompletion has crossed the limits. Throw error
     verify(shuffle, times(0)).reportException(any(Throwable.class));
@@ -487,15 +487,15 @@
     for (int i = 0; i < 64; i++) {
       InputAttemptIdentifier inputAttemptIdentifier =
           new InputAttemptIdentifier(i, 0, "attempt_");
+      InputAttemptFetchFailure failure =
+          InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier);
 
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i %
-          totalProducerNodes), 10000, i, 1), false, true, false);
-
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i %
-          totalProducerNodes), 10000, i, 1), false, true, false);
-
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i %
-          totalProducerNodes), 10000, i, 1), false, true, false);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
+          false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
+          false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
+          false, true);
 
       MapOutput mapOutput = MapOutput
           .createMemoryMapOutput(inputAttemptIdentifier, mock(FetchedInputAllocatorOrderedGrouped.class),
@@ -518,8 +518,8 @@
     //1 fails (last fetch)
     InputAttemptIdentifier inputAttemptIdentifier =
         new InputAttemptIdentifier(319, 0, "attempt_");
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (319 % totalProducerNodes),
-        10000, 319, 1), false, true, false);
+    scheduler.copyFailed(new InputAttemptFetchFailure(inputAttemptIdentifier),
+        new MapHost("host" + (319 % totalProducerNodes), 10000, 319, 1), false, true);
 
     //stall the shuffle (but within limits)
     scheduler.lastProgressTime = System.currentTimeMillis() - 100000;
@@ -527,15 +527,13 @@
     assertEquals(scheduler.remainingMaps.get(), 1);
 
     //Retry for 3 more times
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (319 %
-        totalProducerNodes),
-        10000, 319, 1), false, true, false);
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (319 %
-        totalProducerNodes),
-        10000, 319, 1), false, true, false);
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (319 %
-        totalProducerNodes),
-        10000, 319, 1), false, true, false);
+    InputAttemptFetchFailure failure = InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier);
+    scheduler.copyFailed(failure, new MapHost("host" + (319 % totalProducerNodes), 10000, 319, 1),
+        false, true);
+    scheduler.copyFailed(failure, new MapHost("host" + (319 % totalProducerNodes), 10000, 319, 1),
+        false, true);
+    scheduler.copyFailed(failure, new MapHost("host" + (319 % totalProducerNodes), 10000, 319, 1),
+        false, true);
 
     // failedShufflesSinceLastCompletion has crossed the limits. 20% of other nodes had failures as
     // well. However, it has failed only in one host. So this should proceed
@@ -544,9 +542,8 @@
 
     //stall the shuffle (but within limits)
     scheduler.lastProgressTime = System.currentTimeMillis() - 300000;
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (319 %
-        totalProducerNodes),
-        10000, 319, 1), false, true, false);
+    scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier),
+        new MapHost("host" + (319 % totalProducerNodes), 10000, 319, 1), false, true);
     verify(shuffle, times(1)).reportException(any(Throwable.class));
 
   }
@@ -592,8 +589,9 @@
     //1 fails (last fetch)
     InputAttemptIdentifier inputAttemptIdentifier =
         new InputAttemptIdentifier(318, 0, "attempt_");
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (318 % totalProducerNodes),
-        10000, 318, 1), false, true, false);
+    InputAttemptFetchFailure failure = new InputAttemptFetchFailure(inputAttemptIdentifier);
+    scheduler.copyFailed(failure, new MapHost("host" + (318 % totalProducerNodes),
+        10000, 318, 1), false, true);
 
     //stall the shuffle
     scheduler.lastProgressTime = System.currentTimeMillis() - 1000000;
@@ -601,15 +599,12 @@
     assertEquals(scheduler.remainingMaps.get(), 1);
 
     //Retry for 3 more times
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (318 %
-        totalProducerNodes),
-        10000, 318, 1), false, true, false);
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (318 %
-        totalProducerNodes),
-        10000, 318, 1), false, true, false);
-    scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (318 %
-        totalProducerNodes),
-        10000, 318, 1), false, true, false);
+    scheduler.copyFailed(failure, new MapHost("host" + (318 % totalProducerNodes), 10000, 318, 1),
+        false, true);
+    scheduler.copyFailed(failure, new MapHost("host" + (318 % totalProducerNodes), 10000, 318, 1),
+        false, true);
+    scheduler.copyFailed(failure, new MapHost("host" + (318 % totalProducerNodes), 10000, 318, 1),
+        false, true);
 
     //Shuffle has not received the events completely. So do not bail out yet.
     verify(shuffle, times(0)).reportException(any(Throwable.class));
@@ -672,8 +667,8 @@
     for (int i = 10; i < 15; i++) {
       InputAttemptIdentifier inputAttemptIdentifier =
           new InputAttemptIdentifier(i, 0, "attempt_");
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
+      scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier),
+          new MapHost("host" + (i % totalProducerNodes), 10000, i, 1), false, true);
     }
 
     assertTrue(scheduler.failureCounts.size() >= 5);
@@ -686,10 +681,10 @@
     for (int i = 10; i < 15; i++) {
       InputAttemptIdentifier inputAttemptIdentifier =
           new InputAttemptIdentifier(i, 0, "attempt_");
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
-      scheduler.copyFailed(inputAttemptIdentifier, new MapHost("host" + (i % totalProducerNodes),
-          10000, i, 1), false, true, false);
+      scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier),
+          new MapHost("host" + (i % totalProducerNodes), 10000, i, 1), false, true);
+      scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier),
+          new MapHost("host" + (i % totalProducerNodes), 10000, i, 1), false, true);
     }
 
     boolean checkFailedFetchSinceLastCompletion = conf.getBoolean
@@ -749,18 +744,15 @@
     for (int i = 100; i < 199; i++) {
       InputAttemptIdentifier inputAttemptIdentifier =
           new InputAttemptIdentifier(i, 0, "attempt_");
-      scheduler.copyFailed(inputAttemptIdentifier,
-          new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
-          false, true, false);
-      scheduler.copyFailed(inputAttemptIdentifier,
-          new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
-          false, true, false);
-      scheduler.copyFailed(inputAttemptIdentifier,
-          new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
-          false, true, false);
-      scheduler.copyFailed(inputAttemptIdentifier,
-          new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
-          false, true, false);
+      InputAttemptFetchFailure failure = InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
+          false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
+          false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
+          false, true);
+      scheduler.copyFailed(failure, new MapHost("host" + (i % totalProducerNodes), 10000, i, 1),
+          false, true);
     }
 
     verify(shuffle, atLeast(1)).reportException(any(Throwable.class));
@@ -799,7 +791,8 @@
     MapHost mapHost = scheduler.pendingHosts.iterator().next();
 
     //Fails to pull from host0. host0 should be added to penalties
-    scheduler.copyFailed(inputAttemptIdentifier, mapHost, false, true, false);
+    scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(inputAttemptIdentifier), mapHost,
+        false, true);
 
     //Should not get host, as it is added to penalty loop
     MapHost host = scheduler.getHost();
@@ -993,7 +986,8 @@
     }
 
     for (int i = 0; i < 10; i++) {
-      scheduler.copyFailed(identifiers[0], mapHosts[0], false, false, false);
+      scheduler.copyFailed(InputAttemptFetchFailure.fromAttempt(identifiers[0]), mapHosts[0], false,
+          false);
     }
     ShuffleScheduler.Penalty[] penaltyArray = new ShuffleScheduler.Penalty[scheduler.getPenalties().size()];
     scheduler.getPenalties().toArray(penaltyArray);
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/testutils/RuntimeTestUtils.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/testutils/RuntimeTestUtils.java
new file mode 100644
index 0000000..0885178
--- /dev/null
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/testutils/RuntimeTestUtils.java
@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.tez.runtime.library.testutils;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.ShuffleHeader;
+
+public final class RuntimeTestUtils {
+
+  private RuntimeTestUtils() {
+  }
+
+  public static DataInputStream shuffleHeaderToDataInput(ShuffleHeader header) throws IOException {
+    ByteArrayOutputStream byteOutput = new ByteArrayOutputStream(1000);
+    DataOutputStream output = new DataOutputStream(byteOutput);
+    header.write(output);
+
+    InputStream inputStream = new ByteArrayInputStream(byteOutput.toByteArray());
+    DataInputStream input = new DataInputStream(inputStream);
+
+    return input;
+  }
+}