MAPREDUCE-7048. Uber AM can crash due to unknown task in statusUpdate. Contributed by Peter Bacsko
This commit is contained in:
parent
35c17351ca
commit
87e2570a14
@ -200,6 +200,7 @@ static synchronized String getOutputName(int partition) {
|
||||
protected SecretKey shuffleSecret;
|
||||
protected GcTimeUpdater gcUpdater;
|
||||
final AtomicBoolean mustPreempt = new AtomicBoolean(false);
|
||||
private boolean uberized = false;
|
||||
|
||||
////////////////////////////////////////////
|
||||
// Constructors
|
||||
@ -855,9 +856,6 @@ public void run() {
|
||||
long taskProgressInterval = MRJobConfUtil.
|
||||
getTaskProgressReportInterval(conf);
|
||||
|
||||
boolean uberized = conf.getBoolean("mapreduce.task.uberized",
|
||||
false);
|
||||
|
||||
while (!taskDone.get()) {
|
||||
synchronized (lock) {
|
||||
done = false;
|
||||
@ -1301,11 +1299,17 @@ boolean isCommitRequired() throws IOException {
|
||||
public void statusUpdate(TaskUmbilicalProtocol umbilical)
|
||||
throws IOException {
|
||||
int retries = MAX_RETRIES;
|
||||
|
||||
while (true) {
|
||||
try {
|
||||
if (!umbilical.statusUpdate(getTaskID(), taskStatus).getTaskFound()) {
|
||||
LOG.warn("Parent died. Exiting "+taskId);
|
||||
System.exit(66);
|
||||
if (uberized) {
|
||||
LOG.warn("Task no longer available: " + taskId);
|
||||
break;
|
||||
} else {
|
||||
LOG.warn("Parent died. Exiting " + taskId);
|
||||
ExitUtil.terminate(66);
|
||||
}
|
||||
}
|
||||
taskStatus.clearStatus();
|
||||
return;
|
||||
@ -1518,6 +1522,8 @@ public void setConf(Configuration conf) {
|
||||
NetUtils.addStaticResolution(name, resolvedName);
|
||||
}
|
||||
}
|
||||
|
||||
uberized = conf.getBoolean("mapreduce.task.uberized", false);
|
||||
}
|
||||
|
||||
public Configuration getConf() {
|
||||
|
@ -0,0 +1,89 @@
|
||||
/**
|
||||
* 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.hadoop.mapred;
|
||||
|
||||
import static org.mockito.Matchers.any;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.util.ExitUtil;
|
||||
import org.apache.hadoop.util.ExitUtil.ExitException;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.runners.MockitoJUnitRunner;
|
||||
|
||||
@RunWith(MockitoJUnitRunner.class)
|
||||
public class TestTask {
|
||||
@Mock
|
||||
private TaskUmbilicalProtocol umbilical;
|
||||
|
||||
@Mock
|
||||
private AMFeedback feedback;
|
||||
|
||||
private Task task;
|
||||
|
||||
@Before
|
||||
public void setup() {
|
||||
task = new StubTask();
|
||||
ExitUtil.disableSystemExit();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testStatusUpdateDoesNotExitInUberMode() throws Exception {
|
||||
setupTest(true);
|
||||
|
||||
task.statusUpdate(umbilical);
|
||||
}
|
||||
|
||||
@Test(expected = ExitException.class)
|
||||
public void testStatusUpdateExitsInNonUberMode() throws Exception {
|
||||
setupTest(false);
|
||||
|
||||
task.statusUpdate(umbilical);
|
||||
}
|
||||
|
||||
private void setupTest(boolean uberized)
|
||||
throws IOException, InterruptedException {
|
||||
Configuration conf = new Configuration(false);
|
||||
conf.setBoolean("mapreduce.task.uberized", uberized);
|
||||
task.setConf(conf);
|
||||
when(umbilical.statusUpdate(any(TaskAttemptID.class),
|
||||
any(TaskStatus.class))).thenReturn(feedback);
|
||||
|
||||
// to avoid possible infinite loop
|
||||
when(feedback.getTaskFound()).thenReturn(false, true);
|
||||
}
|
||||
|
||||
public class StubTask extends Task {
|
||||
@Override
|
||||
public void run(JobConf job, TaskUmbilicalProtocol umbilical)
|
||||
throws IOException, ClassNotFoundException, InterruptedException {
|
||||
// nop
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isMapTask() {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
Loading…
Reference in New Issue
Block a user