Merge r1502914 from trunk to branch-2 for YARN-763. AMRMClientAsync should stop heartbeating after receiving shutdown from RM (Xuan Gong via bikas)
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1502916 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
c1b0106711
commit
469e65c3aa
|
@ -669,6 +669,9 @@ Release 2.1.0-beta - 2013-07-02
|
||||||
YARN-541. getAllocatedContainers() is not returning all the allocated
|
YARN-541. getAllocatedContainers() is not returning all the allocated
|
||||||
containers (bikas)
|
containers (bikas)
|
||||||
|
|
||||||
|
YARN-763. AMRMClientAsync should stop heartbeating after receiving
|
||||||
|
shutdown from RM (Xuan Gong via bikas)
|
||||||
|
|
||||||
BREAKDOWN OF HADOOP-8562 SUBTASKS AND RELATED JIRAS
|
BREAKDOWN OF HADOOP-8562 SUBTASKS AND RELATED JIRAS
|
||||||
|
|
||||||
YARN-158. Yarn creating package-info.java must not depend on sh.
|
YARN-158. Yarn creating package-info.java must not depend on sh.
|
||||||
|
|
|
@ -31,6 +31,7 @@ import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||||
import org.apache.hadoop.conf.Configuration;
|
import org.apache.hadoop.conf.Configuration;
|
||||||
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
|
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
|
||||||
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
|
import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
|
||||||
|
import org.apache.hadoop.yarn.api.records.AMCommand;
|
||||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||||
import org.apache.hadoop.yarn.api.records.Container;
|
import org.apache.hadoop.yarn.api.records.Container;
|
||||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||||
|
@ -92,6 +93,7 @@ extends AMRMClientAsync<T> {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected void serviceStart() throws Exception {
|
protected void serviceStart() throws Exception {
|
||||||
|
handlerThread.setDaemon(true);
|
||||||
handlerThread.start();
|
handlerThread.start();
|
||||||
client.start();
|
client.start();
|
||||||
super.serviceStart();
|
super.serviceStart();
|
||||||
|
@ -99,27 +101,19 @@ extends AMRMClientAsync<T> {
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Tells the heartbeat and handler threads to stop and waits for them to
|
* Tells the heartbeat and handler threads to stop and waits for them to
|
||||||
* terminate. Calling this method from the callback handler thread would cause
|
* terminate.
|
||||||
* deadlock, and thus should be avoided.
|
|
||||||
*/
|
*/
|
||||||
@Override
|
@Override
|
||||||
protected void serviceStop() throws Exception {
|
protected void serviceStop() throws Exception {
|
||||||
if (Thread.currentThread() == handlerThread) {
|
|
||||||
throw new YarnRuntimeException("Cannot call stop from callback handler thread!");
|
|
||||||
}
|
|
||||||
keepRunning = false;
|
keepRunning = false;
|
||||||
|
heartbeatThread.interrupt();
|
||||||
try {
|
try {
|
||||||
heartbeatThread.join();
|
heartbeatThread.join();
|
||||||
} catch (InterruptedException ex) {
|
} catch (InterruptedException ex) {
|
||||||
LOG.error("Error joining with heartbeat thread", ex);
|
LOG.error("Error joining with heartbeat thread", ex);
|
||||||
}
|
}
|
||||||
client.stop();
|
client.stop();
|
||||||
try {
|
|
||||||
handlerThread.interrupt();
|
handlerThread.interrupt();
|
||||||
handlerThread.join();
|
|
||||||
} catch (InterruptedException ex) {
|
|
||||||
LOG.error("Error joining with hander thread", ex);
|
|
||||||
}
|
|
||||||
super.serviceStop();
|
super.serviceStop();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -248,6 +242,10 @@ extends AMRMClientAsync<T> {
|
||||||
while (true) {
|
while (true) {
|
||||||
try {
|
try {
|
||||||
responseQueue.put(response);
|
responseQueue.put(response);
|
||||||
|
if (response.getAMCommand() == AMCommand.AM_RESYNC
|
||||||
|
|| response.getAMCommand() == AMCommand.AM_SHUTDOWN) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
break;
|
break;
|
||||||
} catch (InterruptedException ex) {
|
} catch (InterruptedException ex) {
|
||||||
LOG.info("Interrupted while waiting to put on response queue", ex);
|
LOG.info("Interrupted while waiting to put on response queue", ex);
|
||||||
|
@ -285,24 +283,18 @@ extends AMRMClientAsync<T> {
|
||||||
}
|
}
|
||||||
|
|
||||||
if (response.getAMCommand() != null) {
|
if (response.getAMCommand() != null) {
|
||||||
boolean stop = false;
|
|
||||||
switch(response.getAMCommand()) {
|
switch(response.getAMCommand()) {
|
||||||
case AM_RESYNC:
|
case AM_RESYNC:
|
||||||
case AM_SHUTDOWN:
|
case AM_SHUTDOWN:
|
||||||
handler.onShutdownRequest();
|
handler.onShutdownRequest();
|
||||||
LOG.info("Shutdown requested. Stopping callback.");
|
LOG.info("Shutdown requested. Stopping callback.");
|
||||||
stop = true;
|
return;
|
||||||
break;
|
|
||||||
default:
|
default:
|
||||||
String msg =
|
String msg =
|
||||||
"Unhandled value of AMCommand: " + response.getAMCommand();
|
"Unhandled value of AMCommand: " + response.getAMCommand();
|
||||||
LOG.error(msg);
|
LOG.error(msg);
|
||||||
throw new YarnRuntimeException(msg);
|
throw new YarnRuntimeException(msg);
|
||||||
}
|
}
|
||||||
if(stop) {
|
|
||||||
// should probably stop heartbeating also YARN-763
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
List<NodeReport> updatedNodes = response.getUpdatedNodes();
|
List<NodeReport> updatedNodes = response.getUpdatedNodes();
|
||||||
if (!updatedNodes.isEmpty()) {
|
if (!updatedNodes.isEmpty()) {
|
||||||
|
|
|
@ -23,7 +23,10 @@ import static org.mockito.Matchers.anyInt;
|
||||||
import static org.mockito.Matchers.anyString;
|
import static org.mockito.Matchers.anyString;
|
||||||
import static org.mockito.Mockito.mock;
|
import static org.mockito.Mockito.mock;
|
||||||
import static org.mockito.Mockito.when;
|
import static org.mockito.Mockito.when;
|
||||||
|
import static org.mockito.Mockito.times;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
|
||||||
|
import java.io.IOException;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.Arrays;
|
import java.util.Arrays;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
@ -218,6 +221,65 @@ public class TestAMRMClientAsync {
|
||||||
Assert.assertTrue(callbackHandler.callbackCount == 0);
|
Assert.assertTrue(callbackHandler.callbackCount == 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test (timeout = 10000)
|
||||||
|
public void testAMRMClientAsyncShutDown() throws Exception {
|
||||||
|
Configuration conf = new Configuration();
|
||||||
|
TestCallbackHandler callbackHandler = new TestCallbackHandler();
|
||||||
|
@SuppressWarnings("unchecked")
|
||||||
|
AMRMClient<ContainerRequest> client = mock(AMRMClientImpl.class);
|
||||||
|
|
||||||
|
final AllocateResponse shutDownResponse = createAllocateResponse(
|
||||||
|
new ArrayList<ContainerStatus>(), new ArrayList<Container>(), null);
|
||||||
|
shutDownResponse.setAMCommand(AMCommand.AM_SHUTDOWN);
|
||||||
|
when(client.allocate(anyFloat())).thenReturn(shutDownResponse);
|
||||||
|
|
||||||
|
AMRMClientAsync<ContainerRequest> asyncClient =
|
||||||
|
AMRMClientAsync.createAMRMClientAsync(client, 10, callbackHandler);
|
||||||
|
asyncClient.init(conf);
|
||||||
|
asyncClient.start();
|
||||||
|
|
||||||
|
asyncClient.registerApplicationMaster("localhost", 1234, null);
|
||||||
|
|
||||||
|
Thread.sleep(50);
|
||||||
|
|
||||||
|
verify(client, times(1)).allocate(anyFloat());
|
||||||
|
asyncClient.stop();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test (timeout = 5000)
|
||||||
|
public void testCallAMRMClientAsyncStopFromCallbackHandler()
|
||||||
|
throws YarnException, IOException, InterruptedException {
|
||||||
|
Configuration conf = new Configuration();
|
||||||
|
TestCallbackHandler2 callbackHandler = new TestCallbackHandler2();
|
||||||
|
@SuppressWarnings("unchecked")
|
||||||
|
AMRMClient<ContainerRequest> client = mock(AMRMClientImpl.class);
|
||||||
|
|
||||||
|
List<ContainerStatus> completed = Arrays.asList(
|
||||||
|
ContainerStatus.newInstance(newContainerId(0, 0, 0, 0),
|
||||||
|
ContainerState.COMPLETE, "", 0));
|
||||||
|
final AllocateResponse response = createAllocateResponse(completed,
|
||||||
|
new ArrayList<Container>(), null);
|
||||||
|
|
||||||
|
when(client.allocate(anyFloat())).thenReturn(response);
|
||||||
|
|
||||||
|
AMRMClientAsync<ContainerRequest> asyncClient =
|
||||||
|
AMRMClientAsync.createAMRMClientAsync(client, 20, callbackHandler);
|
||||||
|
callbackHandler.registerAsyncClient(asyncClient);
|
||||||
|
asyncClient.init(conf);
|
||||||
|
asyncClient.start();
|
||||||
|
|
||||||
|
synchronized (callbackHandler.notifier) {
|
||||||
|
asyncClient.registerApplicationMaster("localhost", 1234, null);
|
||||||
|
while(callbackHandler.stop == false) {
|
||||||
|
try {
|
||||||
|
callbackHandler.notifier.wait();
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
e.printStackTrace();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private AllocateResponse createAllocateResponse(
|
private AllocateResponse createAllocateResponse(
|
||||||
List<ContainerStatus> completed, List<Container> allocated,
|
List<ContainerStatus> completed, List<Container> allocated,
|
||||||
List<NMToken> nmTokens) {
|
List<NMToken> nmTokens) {
|
||||||
|
@ -323,4 +385,41 @@ public class TestAMRMClientAsync {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private class TestCallbackHandler2 implements AMRMClientAsync.CallbackHandler {
|
||||||
|
Object notifier = new Object();
|
||||||
|
@SuppressWarnings("rawtypes")
|
||||||
|
AMRMClientAsync asynClient;
|
||||||
|
boolean stop = false;
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onContainersCompleted(List<ContainerStatus> statuses) {}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onContainersAllocated(List<Container> containers) {}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onShutdownRequest() {}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onNodesUpdated(List<NodeReport> updatedNodes) {}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public float getProgress() {
|
||||||
|
asynClient.stop();
|
||||||
|
stop = true;
|
||||||
|
synchronized (notifier) {
|
||||||
|
notifier.notifyAll();
|
||||||
|
}
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onError(Exception e) {}
|
||||||
|
|
||||||
|
public void registerAsyncClient(
|
||||||
|
AMRMClientAsync<ContainerRequest> asyncClient) {
|
||||||
|
this.asynClient = asyncClient;
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue