You are viewing a plain text version of this content. The canonical link for it is here.
Posted to mapreduce-commits@hadoop.apache.org by jl...@apache.org on 2013/08/02 02:07:52 UTC
svn commit: r1509486 - in
/hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project: ./
hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/
hadoop-mapreduce-client/hadoop-mapreduce-client-c...
Author: jlowe
Date: Fri Aug 2 00:07:52 2013
New Revision: 1509486
URL: http://svn.apache.org/r1509486
Log:
svn merge -c 1507104 FIXES: MAPREDUCE-5251. Reducer should not implicate map attempt if it has insufficient space to fetch map output. Contributed by Ashwin Shankar
Modified:
hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/CHANGES.txt
hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java
hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/ShuffleSchedulerImpl.java
hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestFetcher.java
Modified: hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/CHANGES.txt?rev=1509486&r1=1509485&r2=1509486&view=diff
==============================================================================
--- hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/CHANGES.txt (original)
+++ hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/CHANGES.txt Fri Aug 2 00:07:52 2013
@@ -21,6 +21,9 @@ Release 2.1.1-beta - UNRELEASED
MAPREDUCE-5428. HistoryFileManager doesn't stop threads when service is
stopped (Karthik Kambatla via jlowe)
+ MAPREDUCE-5251. Reducer should not implicate map attempt if it has
+ insufficient space to fetch map output (Ashwin Shankar via jlowe)
+
Release 2.1.0-beta - 2013-08-06
INCOMPATIBLE CHANGES
Modified: hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java?rev=1509486&r1=1509485&r2=1509486&view=diff
==============================================================================
--- hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java (original)
+++ hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/Fetcher.java Fri Aug 2 00:07:52 2013
@@ -407,7 +407,14 @@ class Fetcher<K,V> extends Thread {
}
// Get the location for the map output - either in-memory or on-disk
- mapOutput = merger.reserve(mapId, decompressedLength, id);
+ try {
+ mapOutput = merger.reserve(mapId, decompressedLength, id);
+ } catch (IOException ioe) {
+ // kill this reduce attempt
+ ioErrs.increment(1);
+ scheduler.reportLocalError(ioe);
+ return EMPTY_ATTEMPT_ID_ARRAY;
+ }
// Check if we can shuffle *now* ...
if (mapOutput == null) {
Modified: hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/ShuffleSchedulerImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/ShuffleSchedulerImpl.java?rev=1509486&r1=1509485&r2=1509486&view=diff
==============================================================================
--- hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/ShuffleSchedulerImpl.java (original)
+++ hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/ShuffleSchedulerImpl.java Fri Aug 2 00:07:52 2013
@@ -19,7 +19,9 @@ package org.apache.hadoop.mapreduce.task
import java.io.IOException;
+import java.net.InetAddress;
import java.net.URI;
+import java.net.UnknownHostException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.HashMap;
@@ -252,6 +254,16 @@ public class ShuffleSchedulerImpl<K,V> i
failedShuffleCounter.increment(1);
}
+
+ public void reportLocalError(IOException ioe) {
+ try {
+ LOG.error("Shuffle failed : local error on this node: "
+ + InetAddress.getLocalHost());
+ } catch (UnknownHostException e) {
+ LOG.error("Shuffle failed : local error on this node");
+ }
+ reporter.reportException(ioe);
+ }
// Notify the JobTracker
// after every read error, if 'reportReadErrorImmediately' is true or
Modified: hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestFetcher.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestFetcher.java?rev=1509486&r1=1509485&r2=1509486&view=diff
==============================================================================
--- hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestFetcher.java (original)
+++ hadoop/common/branches/branch-2.1-beta/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/task/reduce/TestFetcher.java Fri Aug 2 00:07:52 2013
@@ -58,6 +58,7 @@ import org.apache.hadoop.mapred.Reporter
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
import org.apache.hadoop.mapreduce.security.token.JobTokenSecretManager;
+import org.apache.hadoop.util.DiskChecker.DiskErrorException;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
@@ -114,6 +115,36 @@ public class TestFetcher {
LOG.info("<<<< " + name.getMethodName());
}
+ @Test
+ public void testReduceOutOfDiskSpace() throws Throwable {
+ LOG.info("testReduceOutOfDiskSpace");
+
+ Fetcher<Text,Text> underTest = new FakeFetcher<Text,Text>(job, id, ss, mm,
+ r, metrics, except, key, connection);
+
+ String replyHash = SecureShuffleUtils.generateHash(encHash.getBytes(), key);
+ ShuffleHeader header = new ShuffleHeader(map1ID.toString(), 10, 10, 1);
+ ByteArrayOutputStream bout = new ByteArrayOutputStream();
+ header.write(new DataOutputStream(bout));
+
+ ByteArrayInputStream in = new ByteArrayInputStream(bout.toByteArray());
+
+ when(connection.getResponseCode()).thenReturn(200);
+ when(connection.getHeaderField(ShuffleHeader.HTTP_HEADER_NAME))
+ .thenReturn(ShuffleHeader.DEFAULT_HTTP_HEADER_NAME);
+ when(connection.getHeaderField(ShuffleHeader.HTTP_HEADER_VERSION))
+ .thenReturn(ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION);
+ when(connection.getHeaderField(SecureShuffleUtils.HTTP_HEADER_REPLY_URL_HASH))
+ .thenReturn(replyHash);
+ when(connection.getInputStream()).thenReturn(in);
+
+ when(mm.reserve(any(TaskAttemptID.class), anyLong(), anyInt()))
+ .thenThrow(new DiskErrorException("No disk space available"));
+
+ underTest.copyFromHost(host);
+ verify(ss).reportLocalError(any(IOException.class));
+ }
+
@Test(timeout=30000)
public void testCopyFromHostConnectionTimeout() throws Exception {
when(connection.getInputStream()).thenThrow(