You are viewing a plain text version of this content. The canonical link for it is here.
Posted to yarn-commits@hadoop.apache.org by ac...@apache.org on 2012/12/19 05:34:57 UTC
svn commit: r1423759 [1/2] - in
/hadoop/common/branches/branch-2/hadoop-yarn-project: ./
hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/
hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/...
Author: acmurthy
Date: Wed Dec 19 04:34:54 2012
New Revision: 1423759
URL: http://svn.apache.org/viewvc?rev=1423759&view=rev
Log:
Merge -c 1423758 from trunk to branch-2 to fix YARN-230. RM Restart phase 1 - includes support for saving/restarting all applications on an RM bounce. Contributed by Bikas Saha.
Added:
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ApplicationAttemptStateData.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ApplicationAttemptStateData.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ApplicationStateData.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ApplicationStateData.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ApplicationAttemptStateDataPBImpl.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ApplicationAttemptStateDataPBImpl.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ApplicationStateDataPBImpl.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ApplicationStateDataPBImpl.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/NullRMStateStore.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/NullRMStateStore.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreAppAttemptEvent.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreAppAttemptEvent.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreEvent.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreEvent.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreEventType.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreEventType.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreFactory.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreFactory.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreRemoveAppEvent.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreRemoveAppEvent.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/event/RMAppAttemptStoredEvent.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/event/RMAppAttemptStoredEvent.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMRestart.java
- copied unchanged from r1423758, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMRestart.java
Removed:
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileRMStateStore.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/StoreFactory.java
Modified:
hadoop/common/branches/branch-2/hadoop-yarn-project/CHANGES.txt
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ApplicationSubmissionContextPBImpl.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestGetGroups.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestYarnClient.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContext.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/package-info.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMApp.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptEventType.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptState.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockRM.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestApplicationACLs.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestResourceManager.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/applicationsmanager/MockAsm.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/applicationsmanager/TestASMStateMachine.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/applicationsmanager/TestSchedulerNegotiator.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/MockRMApp.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/TestRMAppAttemptTransitions.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSLeafQueue.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairSchedulerEventLog.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/TestFifoScheduler.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/MiniYARNCluster.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/TestRMNMSecretKeys.java
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/CHANGES.txt?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/CHANGES.txt (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/CHANGES.txt Wed Dec 19 04:34:54 2012
@@ -8,6 +8,9 @@ Release 2.0.3-alpha - Unreleased
YARN-145. Add a Web UI to the fair share scheduler. (Sandy Ryza via tomwhite)
+ YARN-230. RM Restart phase 1 - includes support for saving/restarting all
+ applications on an RM bounce. (Bikas Saha via acmurthy)
+
IMPROVEMENTS
YARN-78. Changed UnManagedAM application to use YarnClient. (Bikas Saha via
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ApplicationSubmissionContextPBImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ApplicationSubmissionContextPBImpl.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ApplicationSubmissionContextPBImpl.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ApplicationSubmissionContextPBImpl.java Wed Dec 19 04:34:54 2012
@@ -210,7 +210,6 @@ implements ApplicationSubmissionContext
@Override
public boolean getUnmanagedAM() {
ApplicationSubmissionContextProtoOrBuilder p = viaProto ? proto : builder;
- //There is a default so cancelTokens should never be null
return p.getUnmanagedAm();
}
@@ -219,7 +218,7 @@ implements ApplicationSubmissionContext
maybeInitBuilder();
builder.setUnmanagedAm(value);
}
-
+
@Override
public boolean getCancelTokensWhenComplete() {
ApplicationSubmissionContextProtoOrBuilder p = viaProto ? proto : builder;
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto Wed Dec 19 04:34:54 2012
@@ -329,3 +329,15 @@ message StringBytesMapProto {
optional bytes value = 2;
}
+////////////////////////////////////////////////////////////////////////
+////// From recovery////////////////////////////////////////////////////
+////////////////////////////////////////////////////////////////////////
+message ApplicationStateDataProto {
+ optional int64 submit_time = 1;
+ optional ApplicationSubmissionContextProto application_submission_context = 2;
+}
+
+message ApplicationAttemptStateDataProto {
+ optional ApplicationAttemptIdProto attemptId = 1;
+ optional ContainerProto master_container = 2;
+}
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestGetGroups.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestGetGroups.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestGetGroups.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestGetGroups.java Wed Dec 19 04:34:54 2012
@@ -28,8 +28,6 @@ import org.apache.hadoop.tools.GetGroups
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
-import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
-import org.apache.hadoop.yarn.server.resourcemanager.recovery.StoreFactory;
import org.apache.hadoop.yarn.service.Service.STATE;
import org.junit.AfterClass;
import org.junit.Before;
@@ -46,8 +44,7 @@ public class TestGetGroups extends GetGr
@BeforeClass
public static void setUpResourceManager() throws IOException, InterruptedException {
conf = new YarnConfiguration();
- RMStateStore store = StoreFactory.getStore(conf);
- resourceManager = new ResourceManager(store) {
+ resourceManager = new ResourceManager() {
@Override
protected void doSecureLogin() throws IOException {
};
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestYarnClient.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestYarnClient.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestYarnClient.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/TestYarnClient.java Wed Dec 19 04:34:54 2012
@@ -34,7 +34,7 @@ public class TestYarnClient {
@Test
public void testClientStop() {
Configuration conf = new Configuration();
- ResourceManager rm = new ResourceManager(null);
+ ResourceManager rm = new ResourceManager();
rm.init(conf);
rm.start();
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java Wed Dec 19 04:34:54 2012
@@ -225,10 +225,12 @@ public class YarnConfiguration extends C
public static final long DELEGATION_TOKEN_MAX_LIFETIME_DEFAULT =
7*24*60*60*1000; // 7 days
+ public static final String RECOVERY_ENABLED = RM_PREFIX + "recovery.enabled";
+ public static final boolean DEFAULT_RM_RECOVERY_ENABLED = false;
/** The class to use as the persistent store.*/
public static final String RM_STORE = RM_PREFIX + "store.class";
-
+
/** The maximum number of completed applications RM keeps. */
public static final String RM_MAX_COMPLETED_APPLICATIONS =
RM_PREFIX + "max-completed-applications";
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml Wed Dec 19 04:34:54 2012
@@ -205,6 +205,13 @@
</property>
<property>
+ <description>Enable RM to recover state after starting. If true, then
+ yarn.resourcemanager.store.class must be specified</description>
+ <name>yarn.resourcemanager.recovery.enabled</name>
+ <value>false</value>
+ </property>
+
+ <property>
<description>The class to use as the persistent store.</description>
<name>yarn.resourcemanager.store.class</name>
</property>
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java Wed Dec 19 04:34:54 2012
@@ -263,6 +263,8 @@ public class ApplicationMasterService ex
} else if (request.getResponseId() + 1 < lastResponse.getResponseId()) {
LOG.error("Invalid responseid from appAttemptId " + appAttemptId);
// Oh damn! Sending reboot isn't enough. RM state is corrupted. TODO:
+ // Reboot is not useful since after AM reboots, it will send register and
+ // get an exception. Might as well throw an exception here.
allocateResponse.setAMResponse(reboot);
return allocateResponse;
}
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java Wed Dec 19 04:34:54 2012
@@ -37,6 +37,7 @@ import org.apache.hadoop.security.UserGr
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
import org.apache.hadoop.security.authorize.PolicyProvider;
import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.ExitUtil;
import org.apache.hadoop.yarn.api.ClientRMProtocol;
import org.apache.hadoop.yarn.api.protocolrecords.GetAllApplicationsRequest;
import org.apache.hadoop.yarn.api.protocolrecords.GetAllApplicationsResponse;
@@ -75,6 +76,7 @@ import org.apache.hadoop.yarn.ipc.YarnRP
import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier;
import org.apache.hadoop.yarn.server.RMDelegationTokenSecretManager;
import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
@@ -254,6 +256,20 @@ public class ClientRMService extends Abs
// So call handle directly and do not send an event.
rmAppManager.handle(new RMAppManagerSubmitEvent(submissionContext, System
.currentTimeMillis()));
+
+ // If recovery is enabled then store the application information in a
+ // blocking call so make sure that RM has stored the information needed
+ // to restart the AM after RM restart without further client communication
+ RMStateStore stateStore = rmContext.getStateStore();
+ LOG.info("Storing Application with id " + applicationId);
+ try {
+ stateStore.storeApplication(rmContext.getRMApps().get(applicationId));
+ } catch (Exception e) {
+ // For HA this exception needs to be handled by giving up
+ // master status if we got fenced
+ LOG.error("Failed to store application:" + applicationId, e);
+ ExitUtil.terminate(1, e);
+ }
LOG.info("Application with id " + applicationId.getId() +
" submitted by user " + user);
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java Wed Dec 19 04:34:54 2012
@@ -20,6 +20,7 @@ package org.apache.hadoop.yarn.server.re
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.LinkedList;
+import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -36,6 +37,10 @@ import org.apache.hadoop.yarn.event.Even
import org.apache.hadoop.yarn.ipc.RPCUtil;
import org.apache.hadoop.yarn.security.client.ClientTokenIdentifier;
import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.Recoverable;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
@@ -48,7 +53,8 @@ import org.apache.hadoop.yarn.server.sec
/**
* This class manages the list of applications for the resource manager.
*/
-public class RMAppManager implements EventHandler<RMAppManagerEvent> {
+public class RMAppManager implements EventHandler<RMAppManagerEvent>,
+ Recoverable {
private static final Log LOG = LogFactory.getLog(RMAppManager.class);
@@ -173,6 +179,10 @@ public class RMAppManager implements Eve
completedApps.add(applicationId);
writeAuditLog(applicationId);
+
+ // application completely done. Remove from state
+ RMStateStore store = rmContext.getStateStore();
+ store.removeApplication(rmContext.getRMApps().get(applicationId));
}
}
@@ -306,6 +316,37 @@ public class RMAppManager implements Eve
}
return credentials;
}
+
+ @Override
+ public void recover(RMState state) throws Exception {
+ RMStateStore store = rmContext.getStateStore();
+ assert store != null;
+ // recover applications
+ Map<ApplicationId, ApplicationState> appStates = state.getApplicationState();
+ LOG.info("Recovering " + appStates.size() + " applications");
+ for(ApplicationState appState : appStates.values()) {
+ // re-submit the application
+ // this is going to send an app start event but since the async dispatcher
+ // has not started that event will be queued until we have completed re
+ // populating the state
+ if(appState.getApplicationSubmissionContext().getUnmanagedAM()) {
+ // do not recover unmanaged applications since current recovery
+ // mechanism of restarting attempts does not work for them.
+ // This will need to be changed in work preserving recovery in which
+ // RM will re-connect with the running AM's instead of restarting them
+ LOG.info("Not recovering unmanaged application " + appState.getAppId());
+ store.removeApplication(appState);
+ } else {
+ LOG.info("Recovering application " + appState.getAppId());
+ submitApplication(appState.getApplicationSubmissionContext(),
+ appState.getSubmitTime());
+ // re-populate attempt information in application
+ RMAppImpl appImpl = (RMAppImpl) rmContext.getRMApps().get(
+ appState.getAppId());
+ appImpl.recover(state);
+ }
+ }
+ }
@Override
public void handle(RMAppManagerEvent event) {
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContext.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContext.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContext.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContext.java Wed Dec 19 04:34:54 2012
@@ -23,6 +23,7 @@ import java.util.concurrent.ConcurrentMa
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.event.Dispatcher;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.AMLivelinessMonitor;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer;
@@ -38,6 +39,8 @@ import org.apache.hadoop.yarn.server.res
public interface RMContext {
Dispatcher getDispatcher();
+
+ RMStateStore getStateStore();
ConcurrentMap<ApplicationId, RMApp> getRMApps();
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java Wed Dec 19 04:34:54 2012
@@ -23,7 +23,10 @@ import java.util.concurrent.ConcurrentMa
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.Dispatcher;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.NullRMStateStore;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.AMLivelinessMonitor;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer;
@@ -33,6 +36,8 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.security.DelegationTokenRenewer;
import org.apache.hadoop.yarn.server.resourcemanager.security.RMContainerTokenSecretManager;
+import com.google.common.annotations.VisibleForTesting;
+
public class RMContextImpl implements RMContext {
private final Dispatcher rmDispatcher;
@@ -48,6 +53,7 @@ public class RMContextImpl implements RM
private AMLivelinessMonitor amLivelinessMonitor;
private AMLivelinessMonitor amFinishingMonitor;
+ private RMStateStore stateStore = null;
private ContainerAllocationExpirer containerAllocationExpirer;
private final DelegationTokenRenewer tokenRenewer;
private final ApplicationTokenSecretManager appTokenSecretManager;
@@ -55,6 +61,7 @@ public class RMContextImpl implements RM
private final ClientToAMTokenSecretManagerInRM clientToAMTokenSecretManager;
public RMContextImpl(Dispatcher rmDispatcher,
+ RMStateStore store,
ContainerAllocationExpirer containerAllocationExpirer,
AMLivelinessMonitor amLivelinessMonitor,
AMLivelinessMonitor amFinishingMonitor,
@@ -63,6 +70,7 @@ public class RMContextImpl implements RM
RMContainerTokenSecretManager containerTokenSecretManager,
ClientToAMTokenSecretManagerInRM clientTokenSecretManager) {
this.rmDispatcher = rmDispatcher;
+ this.stateStore = store;
this.containerAllocationExpirer = containerAllocationExpirer;
this.amLivelinessMonitor = amLivelinessMonitor;
this.amFinishingMonitor = amFinishingMonitor;
@@ -71,11 +79,39 @@ public class RMContextImpl implements RM
this.containerTokenSecretManager = containerTokenSecretManager;
this.clientToAMTokenSecretManager = clientTokenSecretManager;
}
+
+ @VisibleForTesting
+ // helper constructor for tests
+ public RMContextImpl(Dispatcher rmDispatcher,
+ ContainerAllocationExpirer containerAllocationExpirer,
+ AMLivelinessMonitor amLivelinessMonitor,
+ AMLivelinessMonitor amFinishingMonitor,
+ DelegationTokenRenewer tokenRenewer,
+ ApplicationTokenSecretManager appTokenSecretManager,
+ RMContainerTokenSecretManager containerTokenSecretManager,
+ ClientToAMTokenSecretManagerInRM clientTokenSecretManager) {
+ this(rmDispatcher, null, containerAllocationExpirer, amLivelinessMonitor,
+ amFinishingMonitor, tokenRenewer, appTokenSecretManager,
+ containerTokenSecretManager, clientTokenSecretManager);
+ RMStateStore nullStore = new NullRMStateStore();
+ nullStore.setDispatcher(rmDispatcher);
+ try {
+ nullStore.init(new YarnConfiguration());
+ setStateStore(nullStore);
+ } catch (Exception e) {
+ assert false;
+ }
+ }
@Override
public Dispatcher getDispatcher() {
return this.rmDispatcher;
}
+
+ @Override
+ public RMStateStore getStateStore() {
+ return stateStore;
+ }
@Override
public ConcurrentMap<ApplicationId, RMApp> getRMApps() {
@@ -126,4 +162,9 @@ public class RMContextImpl implements RM
public ClientToAMTokenSecretManagerInRM getClientToAMTokenSecretManager() {
return this.clientToAMTokenSecretManager;
}
+
+ @VisibleForTesting
+ public void setStateStore(RMStateStore store) {
+ stateStore = store;
+ }
}
\ No newline at end of file
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java Wed Dec 19 04:34:54 2012
@@ -30,6 +30,7 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.metrics2.source.JvmMetrics;
import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.util.ExitUtil;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.ShutdownHookManager;
import org.apache.hadoop.util.StringUtils;
@@ -45,10 +46,11 @@ import org.apache.hadoop.yarn.event.Even
import org.apache.hadoop.yarn.server.RMDelegationTokenSecretManager;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncherEventType;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.ApplicationMasterLauncher;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.NullRMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.Recoverable;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
-import org.apache.hadoop.yarn.server.resourcemanager.recovery.StoreFactory;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStoreFactory;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
@@ -80,6 +82,8 @@ import org.apache.hadoop.yarn.webapp.Web
import org.apache.hadoop.yarn.webapp.WebApps;
import org.apache.hadoop.yarn.webapp.WebApps.Builder;
+import com.google.common.annotations.VisibleForTesting;
+
/**
* The ResourceManager is the main class that is a set of components.
* "I am the ResourceManager. All your resources are belong to us..."
@@ -119,14 +123,13 @@ public class ResourceManager extends Com
protected RMDelegationTokenSecretManager rmDTSecretManager;
private WebApp webApp;
protected RMContext rmContext;
- private final RMStateStore store;
protected ResourceTrackerService resourceTracker;
+ private boolean recoveryEnabled;
private Configuration conf;
-
- public ResourceManager(RMStateStore store) {
+
+ public ResourceManager() {
super("ResourceManager");
- this.store = store;
}
public RMContext getRMContext() {
@@ -160,12 +163,34 @@ public class ResourceManager extends Com
this.containerTokenSecretManager = createContainerTokenSecretManager(conf);
+ boolean isRecoveryEnabled = conf.getBoolean(
+ YarnConfiguration.RECOVERY_ENABLED,
+ YarnConfiguration.DEFAULT_RM_RECOVERY_ENABLED);
+
+ RMStateStore rmStore = null;
+ if(isRecoveryEnabled) {
+ recoveryEnabled = true;
+ rmStore = RMStateStoreFactory.getStore(conf);
+ } else {
+ recoveryEnabled = false;
+ rmStore = new NullRMStateStore();
+ }
+ try {
+ rmStore.init(conf);
+ rmStore.setDispatcher(rmDispatcher);
+ } catch (Exception e) {
+ // the Exception from stateStore.init() needs to be handled for
+ // HA and we need to give up master status if we got fenced
+ LOG.error("Failed to init state store", e);
+ ExitUtil.terminate(1, e);
+ }
+
this.rmContext =
- new RMContextImpl(this.rmDispatcher,
+ new RMContextImpl(this.rmDispatcher, rmStore,
this.containerAllocationExpirer, amLivelinessMonitor,
amFinishingMonitor, tokenRenewer, this.appTokenSecretManager,
this.containerTokenSecretManager, this.clientToAMSecretManager);
-
+
// Register event handler for NodesListManager
this.nodesListManager = new NodesListManager(this.rmContext);
this.rmDispatcher.register(NodesListManagerEventType.class,
@@ -226,9 +251,15 @@ public class ResourceManager extends Com
addService(applicationMasterLauncher);
new RMNMInfo(this.rmContext, this.scheduler);
-
+
super.init(conf);
}
+
+ @VisibleForTesting
+ protected void setRMStateStore(RMStateStore rmStore) {
+ rmStore.setDispatcher(rmDispatcher);
+ ((RMContextImpl) rmContext).setStateStore(rmStore);
+ }
protected RMContainerTokenSecretManager createContainerTokenSecretManager(
Configuration conf) {
@@ -502,6 +533,19 @@ public class ResourceManager extends Com
this.appTokenSecretManager.start();
this.containerTokenSecretManager.start();
+ if(recoveryEnabled) {
+ try {
+ RMStateStore rmStore = rmContext.getStateStore();
+ RMState state = rmStore.loadState();
+ recover(state);
+ } catch (Exception e) {
+ // the Exception from loadState() needs to be handled for
+ // HA and we need to give up master status if we got fenced
+ LOG.error("Failed to load/recover state", e);
+ ExitUtil.terminate(1, e);
+ }
+ }
+
startWepApp();
DefaultMetricsSystem.initialize("ResourceManager");
JvmMetrics.initSingleton("ResourceManager", null);
@@ -555,6 +599,13 @@ public class ResourceManager extends Com
DefaultMetricsSystem.shutdown();
+ RMStateStore store = rmContext.getStateStore();
+ try {
+ store.close();
+ } catch (Exception e) {
+ LOG.error("Error closing store.", e);
+ }
+
super.stop();
}
@@ -643,6 +694,8 @@ public class ResourceManager extends Com
@Override
public void recover(RMState state) throws Exception {
+ // recover applications
+ rmAppManager.recover(state);
}
public static void main(String argv[]) {
@@ -650,13 +703,11 @@ public class ResourceManager extends Com
StringUtils.startupShutdownMessage(ResourceManager.class, argv, LOG);
try {
Configuration conf = new YarnConfiguration();
- RMStateStore store = StoreFactory.getStore(conf);
- ResourceManager resourceManager = new ResourceManager(store);
+ ResourceManager resourceManager = new ResourceManager();
ShutdownHookManager.get().addShutdownHook(
new CompositeServiceShutdownHook(resourceManager),
SHUTDOWN_HOOK_PRIORITY);
resourceManager.init(conf);
- //resourceManager.recover(store.restore());
resourceManager.start();
} catch (Throwable t) {
LOG.fatal("Error starting ResourceManager", t);
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java Wed Dec 19 04:34:54 2012
@@ -15,10 +15,313 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.hadoop.yarn.server.resourcemanager.recovery;
-public interface RMStateStore {
- public interface RMState {
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationAttemptStateDataPBImpl;
+import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationStateDataPBImpl;
+import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationSubmissionContextPBImpl;
+import org.apache.hadoop.yarn.event.AsyncDispatcher;
+import org.apache.hadoop.yarn.event.Dispatcher;
+import org.apache.hadoop.yarn.event.EventHandler;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptStoredEvent;
+
+@Private
+@Unstable
+/**
+ * Base class to implement storage of ResourceManager state.
+ * Takes care of asynchronous notifications and interfacing with YARN objects.
+ * Real store implementations need to derive from it and implement blocking
+ * store and load methods to actually store and load the state.
+ */
+public abstract class RMStateStore {
+ public static final Log LOG = LogFactory.getLog(RMStateStore.class);
+
+ /**
+ * State of an application attempt
+ */
+ public static class ApplicationAttemptState {
+ final ApplicationAttemptId attemptId;
+ final Container masterContainer;
+
+ public ApplicationAttemptState(ApplicationAttemptId attemptId,
+ Container masterContainer) {
+ this.attemptId = attemptId;
+ this.masterContainer = masterContainer;
+ }
+
+ public Container getMasterContainer() {
+ return masterContainer;
+ }
+ public ApplicationAttemptId getAttemptId() {
+ return attemptId;
+ }
+ }
+
+ /**
+ * State of an application application
+ */
+ public static class ApplicationState {
+ final ApplicationSubmissionContext context;
+ final long submitTime;
+ Map<ApplicationAttemptId, ApplicationAttemptState> attempts =
+ new HashMap<ApplicationAttemptId, ApplicationAttemptState>();
+
+ ApplicationState(long submitTime, ApplicationSubmissionContext context) {
+ this.submitTime = submitTime;
+ this.context = context;
+ }
+
+ public ApplicationId getAppId() {
+ return context.getApplicationId();
+ }
+ public long getSubmitTime() {
+ return submitTime;
+ }
+ public int getAttemptCount() {
+ return attempts.size();
+ }
+ public ApplicationSubmissionContext getApplicationSubmissionContext() {
+ return context;
+ }
+ public ApplicationAttemptState getAttempt(ApplicationAttemptId attemptId) {
+ return attempts.get(attemptId);
+ }
+ }
+
+ /**
+ * State of the ResourceManager
+ */
+ public static class RMState {
+ Map<ApplicationId, ApplicationState> appState =
+ new HashMap<ApplicationId, ApplicationState>();
+
+ public Map<ApplicationId, ApplicationState> getApplicationState() {
+ return appState;
+ }
+ }
+
+ private Dispatcher rmDispatcher;
+
+ /**
+ * Dispatcher used to send state operation completion events to
+ * ResourceManager services
+ */
+ public void setDispatcher(Dispatcher dispatcher) {
+ this.rmDispatcher = dispatcher;
+ }
+
+ AsyncDispatcher dispatcher;
+
+ public synchronized void init(Configuration conf) throws Exception{
+ // create async handler
+ dispatcher = new AsyncDispatcher();
+ dispatcher.init(conf);
+ dispatcher.register(RMStateStoreEventType.class,
+ new ForwardingEventHandler());
+ dispatcher.start();
+
+ initInternal(conf);
+ }
+
+ /**
+ * Derived classes initialize themselves using this method.
+ * The base class is initialized and the event dispatcher is ready to use at
+ * this point
+ */
+ protected abstract void initInternal(Configuration conf) throws Exception;
+
+ public synchronized void close() throws Exception {
+ closeInternal();
+ dispatcher.stop();
+ }
+
+ /**
+ * Derived classes close themselves using this method.
+ * The base class will be closed and the event dispatcher will be shutdown
+ * after this
+ */
+ protected abstract void closeInternal() throws Exception;
+
+ /**
+ * Blocking API
+ * The derived class must recover state from the store and return a new
+ * RMState object populated with that state
+ * This must not be called on the dispatcher thread
+ */
+ public abstract RMState loadState() throws Exception;
+
+ /**
+ * Blocking API
+ * ResourceManager services use this to store the application's state
+ * This must not be called on the dispatcher thread
+ */
+ public synchronized void storeApplication(RMApp app) throws Exception {
+ ApplicationSubmissionContext context = app
+ .getApplicationSubmissionContext();
+ assert context instanceof ApplicationSubmissionContextPBImpl;
+
+ ApplicationStateDataPBImpl appStateData = new ApplicationStateDataPBImpl();
+ appStateData.setSubmitTime(app.getSubmitTime());
+ appStateData.setApplicationSubmissionContext(context);
+
+ LOG.info("Storing info for app: " + context.getApplicationId());
+ storeApplicationState(app.getApplicationId().toString(), appStateData);
+ }
+
+ /**
+ * Blocking API
+ * Derived classes must implement this method to store the state of an
+ * application.
+ */
+ protected abstract void storeApplicationState(String appId,
+ ApplicationStateDataPBImpl appStateData)
+ throws Exception;
+
+ @SuppressWarnings("unchecked")
+ /**
+ * Non-blocking API
+ * ResourceManager services call this to store state on an application attempt
+ * This does not block the dispatcher threads
+ * RMAppAttemptStoredEvent will be sent on completion to notify the RMAppAttempt
+ */
+ public synchronized void storeApplicationAttempt(RMAppAttempt appAttempt) {
+ ApplicationAttemptState attemptState = new ApplicationAttemptState(
+ appAttempt.getAppAttemptId(), appAttempt.getMasterContainer());
+ dispatcher.getEventHandler().handle(
+ new RMStateStoreAppAttemptEvent(attemptState));
+ }
+
+ /**
+ * Blocking API
+ * Derived classes must implement this method to store the state of an
+ * application attempt
+ */
+ protected abstract void storeApplicationAttemptState(String attemptId,
+ ApplicationAttemptStateDataPBImpl attemptStateData)
+ throws Exception;
+
+
+ /**
+ * Non-blocking API
+ * ResourceManager services call this to remove an application from the state
+ * store
+ * This does not block the dispatcher threads
+ * There is no notification of completion for this operation.
+ */
+ public synchronized void removeApplication(RMApp app) {
+ ApplicationState appState = new ApplicationState(
+ app.getSubmitTime(), app.getApplicationSubmissionContext());
+ for(RMAppAttempt appAttempt : app.getAppAttempts().values()) {
+ ApplicationAttemptState attemptState = new ApplicationAttemptState(
+ appAttempt.getAppAttemptId(), appAttempt.getMasterContainer());
+ appState.attempts.put(attemptState.getAttemptId(), attemptState);
+ }
+
+ removeApplication(appState);
+ }
+
+ @SuppressWarnings("unchecked")
+ /**
+ * Non-Blocking API
+ */
+ public synchronized void removeApplication(ApplicationState appState) {
+ dispatcher.getEventHandler().handle(new RMStateStoreRemoveAppEvent(appState));
+ }
+
+ /**
+ * Blocking API
+ * Derived classes must implement this method to remove the state of an
+ * application and its attempts
+ */
+ protected abstract void removeApplicationState(ApplicationState appState)
+ throws Exception;
+
+ // Dispatcher related code
+
+ private synchronized void handleStoreEvent(RMStateStoreEvent event) {
+ switch(event.getType()) {
+ case STORE_APP_ATTEMPT:
+ {
+ ApplicationAttemptState attemptState =
+ ((RMStateStoreAppAttemptEvent) event).getAppAttemptState();
+ Exception storedException = null;
+ ApplicationAttemptStateDataPBImpl attemptStateData =
+ new ApplicationAttemptStateDataPBImpl();
+ attemptStateData.setAttemptId(attemptState.getAttemptId());
+ attemptStateData.setMasterContainer(attemptState.getMasterContainer());
+
+ LOG.info("Storing info for attempt: " + attemptState.getAttemptId());
+ try {
+ storeApplicationAttemptState(attemptState.getAttemptId().toString(),
+ attemptStateData);
+ } catch (Exception e) {
+ LOG.error("Error storing appAttempt: "
+ + attemptState.getAttemptId(), e);
+ storedException = e;
+ } finally {
+ notifyDoneStoringApplicationAttempt(attemptState.getAttemptId(),
+ storedException);
+ }
+ }
+ break;
+ case REMOVE_APP:
+ {
+ ApplicationState appState =
+ ((RMStateStoreRemoveAppEvent) event).getAppState();
+ ApplicationId appId = appState.getAppId();
+
+ LOG.info("Removing info for app: " + appId);
+ try {
+ removeApplicationState(appState);
+ } catch (Exception e) {
+ LOG.error("Error removing app: " + appId, e);
+ }
+ }
+ break;
+ default:
+ LOG.error("Unknown RMStateStoreEvent type: " + event.getType());
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ /**
+ * In (@link storeApplicationAttempt}, derived class can call this method to
+ * notify the application attempt about operation completion
+ * @param appAttempt attempt that has been saved
+ */
+ private void notifyDoneStoringApplicationAttempt(ApplicationAttemptId attemptId,
+ Exception storedException) {
+ rmDispatcher.getEventHandler().handle(
+ new RMAppAttemptStoredEvent(attemptId, storedException));
+ }
+
+ /**
+ * EventHandler implementation which forward events to the FSRMStateStore
+ * This hides the EventHandle methods of the store from its public interface
+ */
+ private final class ForwardingEventHandler
+ implements EventHandler<RMStateStoreEvent> {
+
+ @Override
+ public void handle(RMStateStoreEvent event) {
+ handleStoreEvent(event);
+ }
}
+
}
\ No newline at end of file
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/package-info.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/package-info.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/package-info.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/package-info.java Wed Dec 19 04:34:54 2012
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
@InterfaceAudience.Private
@InterfaceStability.Unstable
package org.apache.hadoop.yarn.server.resourcemanager.recovery;
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMApp.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMApp.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMApp.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMApp.java Wed Dec 19 04:34:54 2012
@@ -44,6 +44,12 @@ public interface RMApp extends EventHand
* @return the {@link ApplicationId} for this {@link RMApp}.
*/
ApplicationId getApplicationId();
+
+ /**
+ * The application submission context for this {@link RMApp}
+ * @return the {@link ApplicationSubmissionContext} for this {@link RMApp}
+ */
+ ApplicationSubmissionContext getApplicationSubmissionContext();
/**
* The current state of the {@link RMApp}.
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppImpl.java Wed Dec 19 04:34:54 2012
@@ -50,6 +50,9 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.RMAppManagerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppNodeUpdateEvent.RMAppNodeUpdateType;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.Recoverable;
import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEvent;
@@ -66,7 +69,7 @@ import org.apache.hadoop.yarn.state.Stat
import org.apache.hadoop.yarn.util.BuilderUtils;
import org.apache.hadoop.yarn.util.Records;
-public class RMAppImpl implements RMApp {
+public class RMAppImpl implements RMApp, Recoverable {
private static final Log LOG = LogFactory.getLog(RMAppImpl.class);
private static final String UNAVAILABLE = "N/A";
@@ -243,6 +246,11 @@ public class RMAppImpl implements RMApp
public ApplicationId getApplicationId() {
return this.applicationId;
}
+
+ @Override
+ public ApplicationSubmissionContext getApplicationSubmissionContext() {
+ return this.submissionContext;
+ }
@Override
public FinalApplicationStatus getFinalApplicationStatus() {
@@ -512,9 +520,22 @@ public class RMAppImpl implements RMApp
this.writeLock.unlock();
}
}
+
+ @Override
+ public void recover(RMState state) {
+ ApplicationState appState = state.getApplicationState().get(getApplicationId());
+ LOG.info("Recovering app: " + getApplicationId() + " with " +
+ + appState.getAttemptCount() + " attempts");
+ for(int i=0; i<appState.getAttemptCount(); ++i) {
+ // create attempt
+ createNewAttempt(false);
+ // recover attempt
+ ((RMAppAttemptImpl) currentAttempt).recover(state);
+ }
+ }
@SuppressWarnings("unchecked")
- private void createNewAttempt() {
+ private void createNewAttempt(boolean startAttempt) {
ApplicationAttemptId appAttemptId = Records
.newRecord(ApplicationAttemptId.class);
appAttemptId.setApplicationId(applicationId);
@@ -525,8 +546,10 @@ public class RMAppImpl implements RMApp
submissionContext, conf);
attempts.put(appAttemptId, attempt);
currentAttempt = attempt;
- handler.handle(
- new RMAppAttemptEvent(appAttemptId, RMAppAttemptEventType.START));
+ if(startAttempt) {
+ handler.handle(
+ new RMAppAttemptEvent(appAttemptId, RMAppAttemptEventType.START));
+ }
}
private void processNodeUpdate(RMAppNodeUpdateType type, RMNode node) {
@@ -553,7 +576,7 @@ public class RMAppImpl implements RMApp
private static final class StartAppAttemptTransition extends RMAppTransition {
public void transition(RMAppImpl app, RMAppEvent event) {
- app.createNewAttempt();
+ app.createNewAttempt(true);
};
}
@@ -647,7 +670,7 @@ public class RMAppImpl implements RMApp
msg = "Unmanaged application " + app.getApplicationId()
+ " failed due to " + failedEvent.getDiagnostics()
+ ". Failing the application.";
- } else if (app.attempts.size() == app.maxRetries) {
+ } else if (app.attempts.size() >= app.maxRetries) {
retryApp = false;
msg = "Application " + app.getApplicationId() + " failed "
+ app.maxRetries + " times due to " + failedEvent.getDiagnostics()
@@ -655,7 +678,7 @@ public class RMAppImpl implements RMApp
}
if (retryApp) {
- app.createNewAttempt();
+ app.createNewAttempt(true);
return initialState;
} else {
LOG.info(msg);
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptEventType.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptEventType.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptEventType.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptEventType.java Wed Dec 19 04:34:54 2012
@@ -39,9 +39,15 @@ public enum RMAppAttemptEventType {
CONTAINER_ACQUIRED,
CONTAINER_ALLOCATED,
CONTAINER_FINISHED,
+
+ // Source: RMStateStore
+ ATTEMPT_SAVED,
// Source: Scheduler
APP_REJECTED,
APP_ACCEPTED,
+
+ // Source: RMAttemptImpl.recover
+ RECOVER
}
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java Wed Dec 19 04:34:54 2012
@@ -38,6 +38,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.http.HttpConfig;
+import org.apache.hadoop.util.ExitUtil;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
@@ -57,6 +58,11 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncherEvent;
import org.apache.hadoop.yarn.server.resourcemanager.amlauncher.AMLauncherEventType;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationAttemptState;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.ApplicationState;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.Recoverable;
import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
@@ -69,6 +75,7 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptRegistrationEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptRejectedEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptStatusupdateEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptStoredEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptUnregistrationEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
@@ -85,7 +92,7 @@ import org.apache.hadoop.yarn.state.Stat
import org.apache.hadoop.yarn.util.BuilderUtils;
@SuppressWarnings({"unchecked", "rawtypes"})
-public class RMAppAttemptImpl implements RMAppAttempt {
+public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
private static final Log LOG = LogFactory.getLog(RMAppAttemptImpl.class);
@@ -153,12 +160,15 @@ public class RMAppAttemptImpl implements
.addTransition(RMAppAttemptState.NEW, RMAppAttemptState.FAILED,
RMAppAttemptEventType.REGISTERED,
new UnexpectedAMRegisteredTransition())
+ .addTransition(RMAppAttemptState.NEW, RMAppAttemptState.RECOVERED,
+ RMAppAttemptEventType.RECOVER)
// Transitions from SUBMITTED state
.addTransition(RMAppAttemptState.SUBMITTED, RMAppAttemptState.FAILED,
RMAppAttemptEventType.APP_REJECTED, new AppRejectedTransition())
.addTransition(RMAppAttemptState.SUBMITTED,
- EnumSet.of(RMAppAttemptState.LAUNCHED, RMAppAttemptState.SCHEDULED),
+ EnumSet.of(RMAppAttemptState.LAUNCHED_UNMANAGED_SAVING,
+ RMAppAttemptState.SCHEDULED),
RMAppAttemptEventType.APP_ACCEPTED,
new ScheduleTransition())
.addTransition(RMAppAttemptState.SUBMITTED, RMAppAttemptState.KILLED,
@@ -170,12 +180,42 @@ public class RMAppAttemptImpl implements
// Transitions from SCHEDULED State
.addTransition(RMAppAttemptState.SCHEDULED,
- RMAppAttemptState.ALLOCATED,
+ RMAppAttemptState.ALLOCATED_SAVING,
RMAppAttemptEventType.CONTAINER_ALLOCATED,
new AMContainerAllocatedTransition())
.addTransition(RMAppAttemptState.SCHEDULED, RMAppAttemptState.KILLED,
RMAppAttemptEventType.KILL,
new BaseFinalTransition(RMAppAttemptState.KILLED))
+
+ // Transitions from ALLOCATED_SAVING State
+ .addTransition(RMAppAttemptState.ALLOCATED_SAVING,
+ RMAppAttemptState.ALLOCATED,
+ RMAppAttemptEventType.ATTEMPT_SAVED, new AttemptStoredTransition())
+ .addTransition(RMAppAttemptState.ALLOCATED_SAVING,
+ RMAppAttemptState.ALLOCATED_SAVING,
+ RMAppAttemptEventType.CONTAINER_ACQUIRED,
+ new ContainerAcquiredTransition())
+ // App could be killed by the client. So need to handle this.
+ .addTransition(RMAppAttemptState.ALLOCATED_SAVING,
+ RMAppAttemptState.KILLED,
+ RMAppAttemptEventType.KILL,
+ new BaseFinalTransition(RMAppAttemptState.KILLED))
+
+ // Transitions from LAUNCHED_UNMANAGED_SAVING State
+ .addTransition(RMAppAttemptState.LAUNCHED_UNMANAGED_SAVING,
+ RMAppAttemptState.LAUNCHED,
+ RMAppAttemptEventType.ATTEMPT_SAVED,
+ new UnmanagedAMAttemptSavedTransition())
+ // attempt should not try to register in this state
+ .addTransition(RMAppAttemptState.LAUNCHED_UNMANAGED_SAVING,
+ RMAppAttemptState.FAILED,
+ RMAppAttemptEventType.REGISTERED,
+ new UnexpectedAMRegisteredTransition())
+ // App could be killed by the client. So need to handle this.
+ .addTransition(RMAppAttemptState.LAUNCHED_UNMANAGED_SAVING,
+ RMAppAttemptState.KILLED,
+ RMAppAttemptEventType.KILL,
+ new BaseFinalTransition(RMAppAttemptState.KILLED))
// Transitions from ALLOCATED State
.addTransition(RMAppAttemptState.ALLOCATED,
@@ -279,11 +319,30 @@ public class RMAppAttemptImpl implements
RMAppAttemptEventType.EXPIRE,
RMAppAttemptEventType.REGISTERED,
RMAppAttemptEventType.CONTAINER_ALLOCATED,
+ RMAppAttemptEventType.ATTEMPT_SAVED,
+ RMAppAttemptEventType.CONTAINER_FINISHED,
+ RMAppAttemptEventType.UNREGISTERED,
+ RMAppAttemptEventType.KILL,
+ RMAppAttemptEventType.STATUS_UPDATE))
+
+ // Transitions from RECOVERED State
+ .addTransition(
+ RMAppAttemptState.RECOVERED,
+ RMAppAttemptState.RECOVERED,
+ EnumSet.of(RMAppAttemptEventType.START,
+ RMAppAttemptEventType.APP_ACCEPTED,
+ RMAppAttemptEventType.APP_REJECTED,
+ RMAppAttemptEventType.EXPIRE,
+ RMAppAttemptEventType.LAUNCHED,
+ RMAppAttemptEventType.LAUNCH_FAILED,
+ RMAppAttemptEventType.REGISTERED,
+ RMAppAttemptEventType.CONTAINER_ALLOCATED,
+ RMAppAttemptEventType.CONTAINER_ACQUIRED,
+ RMAppAttemptEventType.ATTEMPT_SAVED,
RMAppAttemptEventType.CONTAINER_FINISHED,
RMAppAttemptEventType.UNREGISTERED,
RMAppAttemptEventType.KILL,
RMAppAttemptEventType.STATUS_UPDATE))
-
.installTopology();
public RMAppAttemptImpl(ApplicationAttemptId appAttemptId,
@@ -318,7 +377,7 @@ public class RMAppAttemptImpl implements
@Override
public ApplicationSubmissionContext getSubmissionContext() {
return this.submissionContext;
- }
+ }
@Override
public FinalApplicationStatus getFinalApplicationStatus() {
@@ -494,6 +553,10 @@ public class RMAppAttemptImpl implements
}
}
+ private void setMasterContainer(Container container) {
+ masterContainer = container;
+ }
+
@Override
public void handle(RMAppAttemptEvent event) {
@@ -561,6 +624,21 @@ public class RMAppAttemptImpl implements
}
}
+ @Override
+ public void recover(RMState state) {
+ ApplicationState appState =
+ state.getApplicationState().get(getAppAttemptId().getApplicationId());
+ ApplicationAttemptState attemptState = appState.getAttempt(getAppAttemptId());
+ assert attemptState != null;
+ setMasterContainer(attemptState.getMasterContainer());
+ LOG.info("Recovered attempt: AppId: " + getAppAttemptId().getApplicationId()
+ + " AttemptId: " + getAppAttemptId()
+ + " MasterContainer: " + masterContainer);
+ setDiagnostics("Attempt recovered after RM restart");
+ handle(new RMAppAttemptEvent(getAppAttemptId(),
+ RMAppAttemptEventType.RECOVER));
+ }
+
private static class BaseTransition implements
SingleArcTransition<RMAppAttemptImpl, RMAppAttemptEvent> {
@@ -625,13 +703,12 @@ public class RMAppAttemptImpl implements
@Override
public RMAppAttemptState transition(RMAppAttemptImpl appAttempt,
RMAppAttemptEvent event) {
-
- // Send the acceptance to the app
- appAttempt.eventHandler.handle(new RMAppEvent(event
- .getApplicationAttemptId().getApplicationId(),
- RMAppEventType.APP_ACCEPTED));
-
if (!appAttempt.submissionContext.getUnmanagedAM()) {
+ // Send the acceptance to the app
+ appAttempt.eventHandler.handle(new RMAppEvent(event
+ .getApplicationAttemptId().getApplicationId(),
+ RMAppEventType.APP_ACCEPTED));
+
// Request a container for the AM.
ResourceRequest request = BuilderUtils.newResourceRequest(
AM_CONTAINER_PRIORITY, "*", appAttempt.submissionContext
@@ -647,35 +724,42 @@ public class RMAppAttemptImpl implements
return RMAppAttemptState.SCHEDULED;
} else {
// RM not allocating container. AM is self launched.
- // Directly go to LAUNCHED state
- // Register with AMLivelinessMonitor
- appAttempt.rmContext.getAMLivelinessMonitor().register(
- appAttempt.applicationAttemptId);
- return RMAppAttemptState.LAUNCHED;
+ RMStateStore store = appAttempt.rmContext.getStateStore();
+ // save state and then go to LAUNCHED state
+ appAttempt.storeAttempt(store);
+ return RMAppAttemptState.LAUNCHED_UNMANAGED_SAVING;
}
}
}
- private static final class AMContainerAllocatedTransition extends BaseTransition {
+ private static final class AMContainerAllocatedTransition
+ extends BaseTransition {
@Override
public void transition(RMAppAttemptImpl appAttempt,
- RMAppAttemptEvent event) {
-
+ RMAppAttemptEvent event) {
// Acquire the AM container from the scheduler.
Allocation amContainerAllocation = appAttempt.scheduler.allocate(
appAttempt.applicationAttemptId, EMPTY_CONTAINER_REQUEST_LIST,
EMPTY_CONTAINER_RELEASE_LIST);
// Set the masterContainer
- appAttempt.masterContainer = amContainerAllocation.getContainers().get(
- 0);
+ appAttempt.setMasterContainer(amContainerAllocation.getContainers().get(
+ 0));
- // Send event to launch the AM Container
- appAttempt.eventHandler.handle(new AMLauncherEvent(
- AMLauncherEventType.LAUNCH, appAttempt));
+ RMStateStore store = appAttempt.rmContext.getStateStore();
+ appAttempt.storeAttempt(store);
}
}
-
+
+ private static final class AttemptStoredTransition extends BaseTransition {
+ @Override
+ public void transition(RMAppAttemptImpl appAttempt,
+ RMAppAttemptEvent event) {
+ appAttempt.checkAttemptStoreError(event);
+ appAttempt.launchAttempt();
+ }
+ }
+
private static class BaseFinalTransition extends BaseTransition {
private final RMAppAttemptState finalAttemptState;
@@ -736,17 +820,34 @@ public class RMAppAttemptImpl implements
}
}
- private static final class AMLaunchedTransition extends BaseTransition {
+ private static class AMLaunchedTransition extends BaseTransition {
@Override
public void transition(RMAppAttemptImpl appAttempt,
- RMAppAttemptEvent event) {
-
+ RMAppAttemptEvent event) {
// Register with AMLivelinessMonitor
- appAttempt.rmContext.getAMLivelinessMonitor().register(
- appAttempt.applicationAttemptId);
-
+ appAttempt.attemptLaunched();
}
}
+
+ private static final class UnmanagedAMAttemptSavedTransition
+ extends AMLaunchedTransition {
+ @Override
+ public void transition(RMAppAttemptImpl appAttempt,
+ RMAppAttemptEvent event) {
+ appAttempt.checkAttemptStoreError(event);
+ // Send the acceptance to the app
+ // Ideally this should have been done when the scheduler accepted the app.
+ // But its here because until the attempt is saved the client should not
+ // launch the unmanaged AM. Client waits for the app status to be accepted
+ // before doing so. So we have to delay the accepted state until we have
+ // completed storing the attempt
+ appAttempt.eventHandler.handle(new RMAppEvent(event
+ .getApplicationAttemptId().getApplicationId(),
+ RMAppEventType.APP_ACCEPTED));
+
+ super.transition(appAttempt, event);
+ }
+ }
private static final class LaunchFailedTransition extends BaseFinalTransition {
@@ -1040,4 +1141,37 @@ public class RMAppAttemptImpl implements
this.readLock.unlock();
}
}
+
+ private void launchAttempt(){
+ // Send event to launch the AM Container
+ eventHandler.handle(new AMLauncherEvent(AMLauncherEventType.LAUNCH, this));
+ }
+
+ private void attemptLaunched() {
+ // Register with AMLivelinessMonitor
+ rmContext.getAMLivelinessMonitor().register(getAppAttemptId());
+ }
+
+ private void checkAttemptStoreError(RMAppAttemptEvent event) {
+ RMAppAttemptStoredEvent storeEvent = (RMAppAttemptStoredEvent) event;
+ if(storeEvent.getStoredException() != null)
+ {
+ // This needs to be handled for HA and give up master status if we got
+ // fenced
+ LOG.error("Failed to store attempt: " + getAppAttemptId(),
+ storeEvent.getStoredException());
+ ExitUtil.terminate(1, storeEvent.getStoredException());
+ }
+ }
+
+ private void storeAttempt(RMStateStore store) {
+ // store attempt data in a non-blocking manner to prevent dispatcher
+ // thread starvation and wait for state to be saved
+ LOG.info("Storing attempt: AppId: " +
+ getAppAttemptId().getApplicationId()
+ + " AttemptId: " +
+ getAppAttemptId()
+ + " MasterContainer: " + masterContainer);
+ store.storeApplicationAttempt(this);
+ }
}
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptState.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptState.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptState.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptState.java Wed Dec 19 04:34:54 2012
@@ -19,6 +19,6 @@
package org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt;
public enum RMAppAttemptState {
- NEW, SUBMITTED, SCHEDULED, ALLOCATED, LAUNCHED, FAILED, RUNNING,
- FINISHING, FINISHED, KILLED,
+ NEW, SUBMITTED, SCHEDULED, ALLOCATED, LAUNCHED, FAILED, RUNNING, FINISHING,
+ FINISHED, KILLED, ALLOCATED_SAVING, LAUNCHED_UNMANAGED_SAVING, RECOVERED
}
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java Wed Dec 19 04:34:54 2012
@@ -47,7 +47,7 @@ public class MockAM {
private volatile int responseId = 0;
private final ApplicationAttemptId attemptId;
private final RMContext context;
- private final AMRMProtocol amRMProtocol;
+ private AMRMProtocol amRMProtocol;
private final List<ResourceRequest> requests = new ArrayList<ResourceRequest>();
private final List<ContainerId> releases = new ArrayList<ContainerId>();
@@ -58,6 +58,10 @@ public class MockAM {
this.amRMProtocol = amRMProtocol;
this.attemptId = attemptId;
}
+
+ void setAMRMProtocol(AMRMProtocol amRMProtocol) {
+ this.amRMProtocol = amRMProtocol;
+ }
public void waitForState(RMAppAttemptState finalState) throws Exception {
RMApp app = context.getRMApps().get(attemptId.getApplicationId());
@@ -66,7 +70,8 @@ public class MockAM {
while (!finalState.equals(attempt.getAppAttemptState())
&& timeoutSecs++ < 20) {
System.out
- .println("AppAttempt State is : " + attempt.getAppAttemptState()
+ .println("AppAttempt : " + attemptId + " State is : "
+ + attempt.getAppAttemptState()
+ " Waiting for state : " + finalState);
Thread.sleep(500);
}
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java?rev=1423759&r1=1423758&r2=1423759&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockNM.java Wed Dec 19 04:34:54 2012
@@ -46,7 +46,7 @@ public class MockNM {
private int responseId;
private NodeId nodeId;
private final int memory;
- private final ResourceTrackerService resourceTracker;
+ private ResourceTrackerService resourceTracker;
private final int httpPort = 2;
private MasterKey currentMasterKey;
@@ -66,6 +66,10 @@ public class MockNM {
public int getHttpPort() {
return httpPort;
}
+
+ void setResourceTrackerService(ResourceTrackerService resourceTracker) {
+ this.resourceTracker = resourceTracker;
+ }
public void containerStatus(Container container) throws Exception {
Map<ApplicationId, List<ContainerStatus>> conts =