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 vi...@apache.org on 2012/02/07 22:59:39 UTC
svn commit: r1241655 - in
/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project: ./
hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/
hadoop-mapreduce-client/hadoop-mapreduce-client-ap...
Author: vinodkv
Date: Tue Feb 7 21:59:39 2012
New Revision: 1241655
URL: http://svn.apache.org/viewvc?rev=1241655&view=rev
Log:
MAPREDUCE-3815. Fixed MR AM to always use hostnames and never IPs when requesting containers so that scheduler can give off data local containers correctly. Contributed by Siddarth Seth.
svn merge --ignore-ancestry -c 1241654 ../../trunk/
Modified:
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskAttempt.java
hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt?rev=1241655&r1=1241654&r2=1241655&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt (original)
+++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/CHANGES.txt Tue Feb 7 21:59:39 2012
@@ -700,6 +700,10 @@ Release 0.23.1 - Unreleased
MAPREDUCE-3436. JobHistory webapp address should use the host configured
in the jobhistory address. (Ahmed Radwan via sseth)
+
+ MAPREDUCE-3815. Fixed MR AM to always use hostnames and never IPs when
+ requesting containers so that scheduler can give off data local containers
+ correctly. (Siddarth Seth via vinodkv)
Release 0.23.0 - 2011-11-01
Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java?rev=1241655&r1=1241654&r2=1241655&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java (original)
+++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java Tue Feb 7 21:59:39 2012
@@ -19,7 +19,9 @@
package org.apache.hadoop.mapreduce.v2.app.job.impl;
import java.io.IOException;
+import java.net.InetAddress;
import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
@@ -32,6 +34,7 @@ import java.util.concurrent.atomic.Atomi
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.regex.Pattern;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -142,7 +145,7 @@ public abstract class TaskAttemptImpl im
protected final JobConf conf;
protected final Path jobFile;
protected final int partition;
- protected final EventHandler eventHandler;
+ protected EventHandler eventHandler;
private final TaskAttemptId attemptId;
private final Clock clock;
private final org.apache.hadoop.mapred.JobID oldJobId;
@@ -1056,7 +1059,7 @@ public abstract class TaskAttemptImpl im
}
}
- private static class RequestContainerTransition implements
+ static class RequestContainerTransition implements
SingleArcTransition<TaskAttemptImpl, TaskAttemptEvent> {
private final boolean rescheduled;
public RequestContainerTransition(boolean rescheduled) {
@@ -1081,14 +1084,44 @@ public abstract class TaskAttemptImpl im
for (String host : taskAttempt.dataLocalHosts) {
racks[i++] = RackResolver.resolve(host).getNetworkLocation();
}
- taskAttempt.eventHandler.handle(
- new ContainerRequestEvent(taskAttempt.attemptId,
- taskAttempt.resourceCapability,
- taskAttempt.dataLocalHosts, racks));
+ taskAttempt.eventHandler.handle(new ContainerRequestEvent(
+ taskAttempt.attemptId, taskAttempt.resourceCapability, taskAttempt
+ .resolveHosts(taskAttempt.dataLocalHosts), racks));
}
}
}
+ protected String[] resolveHosts(String[] src) {
+ String[] result = new String[src.length];
+ for (int i = 0; i < src.length; i++) {
+ if (isIP(src[i])) {
+ result[i] = resolveHost(src[i]);
+ } else {
+ result[i] = src[i];
+ }
+ }
+ return result;
+ }
+
+ protected String resolveHost(String src) {
+ String result = src; // Fallback in case of failure.
+ try {
+ InetAddress addr = InetAddress.getByName(src);
+ result = addr.getHostName();
+ } catch (UnknownHostException e) {
+ LOG.warn("Failed to resolve address: " + src
+ + ". Continuing to use the same.");
+ }
+ return result;
+ }
+
+ private static final Pattern ipPattern = // Pattern for matching ip
+ Pattern.compile("\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}");
+
+ protected boolean isIP(String src) {
+ return ipPattern.matcher(src).matches();
+ }
+
private static class ContainerAssignedTransition implements
SingleArcTransition<TaskAttemptImpl, TaskAttemptEvent> {
@SuppressWarnings({ "unchecked" })
Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskAttempt.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskAttempt.java?rev=1241655&r1=1241654&r2=1241655&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskAttempt.java (original)
+++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskAttempt.java Tue Feb 7 21:59:39 2012
@@ -18,30 +18,54 @@
package org.apache.hadoop.mapreduce.v2.app.job.impl;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import junit.framework.Assert;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MapTaskAttemptImpl;
+import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryEvent;
import org.apache.hadoop.mapreduce.jobhistory.TaskAttemptUnsuccessfulCompletion;
+import org.apache.hadoop.mapreduce.split.JobSplit.TaskSplitMetaInfo;
+import org.apache.hadoop.mapreduce.v2.api.records.JobId;
import org.apache.hadoop.mapreduce.v2.api.records.JobState;
import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId;
import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptReport;
import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptState;
import org.apache.hadoop.mapreduce.v2.api.records.TaskId;
import org.apache.hadoop.mapreduce.v2.api.records.TaskState;
+import org.apache.hadoop.mapreduce.v2.api.records.TaskType;
import org.apache.hadoop.mapreduce.v2.app.AppContext;
import org.apache.hadoop.mapreduce.v2.app.MRApp;
+import org.apache.hadoop.mapreduce.v2.app.TaskAttemptListener;
import org.apache.hadoop.mapreduce.v2.app.job.Job;
import org.apache.hadoop.mapreduce.v2.app.job.Task;
import org.apache.hadoop.mapreduce.v2.app.job.TaskAttempt;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptDiagnosticsUpdateEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEventType;
+import org.apache.hadoop.mapreduce.v2.app.rm.ContainerRequestEvent;
+import org.apache.hadoop.mapreduce.v2.util.MRBuilderUtils;
+import org.apache.hadoop.yarn.Clock;
+import org.apache.hadoop.yarn.SystemClock;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.event.Event;
import org.apache.hadoop.yarn.event.EventHandler;
+import org.apache.hadoop.yarn.util.BuilderUtils;
import org.junit.Test;
+import org.mockito.ArgumentCaptor;
@SuppressWarnings("unchecked")
public class TestTaskAttempt{
@@ -57,6 +81,64 @@ public class TestTaskAttempt{
MRApp app = new FailingAttemptsMRApp(0, 1);
testMRAppHistory(app);
}
+
+ @SuppressWarnings("rawtypes")
+ @Test
+ public void testHostResolveAttempt() throws Exception {
+ TaskAttemptImpl.RequestContainerTransition rct =
+ new TaskAttemptImpl.RequestContainerTransition(false);
+
+ EventHandler eventHandler = mock(EventHandler.class);
+ String[] hosts = new String[3];
+ hosts[0] = "192.168.1.1";
+ hosts[1] = "host2";
+ hosts[2] = "host3";
+ TaskSplitMetaInfo splitInfo =
+ new TaskSplitMetaInfo(hosts, 0, 128 * 1024 * 1024l);
+
+ TaskAttemptImpl mockTaskAttempt =
+ createMapTaskAttemptImplForTest(eventHandler, splitInfo);
+ TaskAttemptImpl spyTa = spy(mockTaskAttempt);
+ when(spyTa.resolveHost(hosts[0])).thenReturn("host1");
+
+ TaskAttemptEvent mockTAEvent = mock(TaskAttemptEvent.class);
+ rct.transition(spyTa, mockTAEvent);
+ verify(spyTa).resolveHost(hosts[0]);
+ ArgumentCaptor<Event> arg = ArgumentCaptor.forClass(Event.class);
+ verify(eventHandler, times(2)).handle(arg.capture());
+ if (!(arg.getAllValues().get(1) instanceof ContainerRequestEvent)) {
+ Assert.fail("Second Event not of type ContainerRequestEvent");
+ }
+ Map<String, Boolean> expected = new HashMap<String, Boolean>();
+ expected.put("host1", true);
+ expected.put("host2", true);
+ expected.put("host3", true);
+ ContainerRequestEvent cre =
+ (ContainerRequestEvent) arg.getAllValues().get(1);
+ String[] requestedHosts = cre.getHosts();
+ for (String h : requestedHosts) {
+ expected.remove(h);
+ }
+ assertEquals(0, expected.size());
+ }
+
+ @SuppressWarnings("rawtypes")
+ private TaskAttemptImpl createMapTaskAttemptImplForTest(
+ EventHandler eventHandler, TaskSplitMetaInfo taskSplitMetaInfo) {
+ ApplicationId appId = BuilderUtils.newApplicationId(1, 1);
+ JobId jobId = MRBuilderUtils.newJobId(appId, 1);
+ TaskId taskId = MRBuilderUtils.newTaskId(jobId, 1, TaskType.MAP);
+ TaskAttemptListener taListener = mock(TaskAttemptListener.class);
+ Path jobFile = mock(Path.class);
+ JobConf jobConf = new JobConf();
+ OutputCommitter outputCommitter = mock(OutputCommitter.class);
+ Clock clock = new SystemClock();
+ TaskAttemptImpl taImpl =
+ new MapTaskAttemptImpl(taskId, 1, eventHandler, jobFile, 1,
+ taskSplitMetaInfo, jobConf, taListener, outputCommitter, null,
+ null, clock);
+ return taImpl;
+ }
private void testMRAppHistory(MRApp app) throws Exception {
Configuration conf = new Configuration();
Modified: hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java?rev=1241655&r1=1241654&r2=1241655&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java (original)
+++ hadoop/common/branches/branch-0.23/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java Tue Feb 7 21:59:39 2012
@@ -59,7 +59,7 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-@SuppressWarnings({ "rawtypes", "deprecation" })
+@SuppressWarnings("rawtypes")
public class TestTaskImpl {
private static final Log LOG = LogFactory.getLog(TestTaskImpl.class);