Fetcher extends Thread { @VisibleForTesting protected void copyFromHost(MapHost host) throws IOException { retryStartTime = 0; List<TaskAttemptID> maps = scheduler.getMapsForHost(host); if (maps.size() == 0) { return; } if(LOG.isDebugEnabled()) { LOG.debug("Fetcher " + id + " going to fetch from " + host + " for: " + maps); } Set<TaskAttemptID> remaining = new HashSet<TaskAttemptID>(maps); URL url = getMapOutputURL(host, maps); DataInputStream input = null; try { input = openShuffleUrl(host, remaining, url); if (input == null) { return; } TaskAttemptID[] failedTasks = null; while (!remaining.isEmpty() && failedTasks == null) { try { failedTasks = copyMapOutput(host, input, remaining, fetchRetryEnabled); } catch (IOException e) { IOUtils.cleanupWithLogger(LOG, input); connection.disconnect(); url = getMapOutputURL(host, remaining); input = openShuffleUrl(host, remaining, url); if (input == null) { return; } } } if(failedTasks != null && failedTasks.length > 0) { LOG.warn("copyMapOutput failed for tasks "+Arrays.toString(failedTasks)); scheduler.hostFailed(host.getHostName()); for(TaskAttemptID left: failedTasks) { scheduler.copyFailed(left, host, true, false); } } if (failedTasks == null && !remaining.isEmpty()) { throw new IOException("server didn't return all expected map outputs: " + remaining.size() + " left."); } input.close(); input = null; } finally { if (input != null) { IOUtils.cleanupWithLogger(LOG, input); input = null; } for (TaskAttemptID left : remaining) { scheduler.putBackKnownMapOutput(host, left); } } } Fetcher(JobConf job, TaskAttemptID reduceId, 
                 ShuffleSchedulerImpl<K,V> scheduler, MergeManager<K,V> merger,
                 Reporter reporter, ShuffleClientMetrics metrics,
                 ExceptionReporter exceptionReporter, SecretKey shuffleKey); @VisibleForTesting  Fetcher(JobConf job, TaskAttemptID reduceId, 
                 ShuffleSchedulerImpl<K,V> scheduler, MergeManager<K,V> merger,
                 Reporter reporter, ShuffleClientMetrics metrics,
                 ExceptionReporter exceptionReporter, SecretKey shuffleKey,
                 int id); void run(); @Override void interrupt(); void shutDown();  }
@Test public void testCopyFromHostConnectionRejected() throws Exception { when(connection.getResponseCode()) .thenReturn(Fetcher.TOO_MANY_REQ_STATUS_CODE); Fetcher<Text, Text> fetcher = new FakeFetcher<>(job, id, ss, mm, r, metrics, except, key, connection); fetcher.copyFromHost(host); Assert.assertEquals("No host failure is expected.", ss.hostFailureCount(host.getHostName()), 0); Assert.assertEquals("No fetch failure is expected.", ss.fetchFailureCount(map1ID), 0); Assert.assertEquals("No fetch failure is expected.", ss.fetchFailureCount(map2ID), 0); verify(ss).penalize(eq(host), anyLong()); verify(ss).putBackKnownMapOutput(any(MapHost.class), eq(map1ID)); verify(ss).putBackKnownMapOutput(any(MapHost.class), eq(map2ID)); }