|
@@ -52,13 +52,13 @@ import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.AMRMTokenS
|
|
|
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData;
|
|
|
import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData;
|
|
|
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
|
|
|
-import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppNewSavedEvent;
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
|
|
|
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
|
|
|
-import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppUpdateSavedEvent;
|
|
|
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEvent;
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEventType;
|
|
|
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
|
|
|
-import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptNewSavedEvent;
|
|
|
-import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAttemptUpdateSavedEvent;
|
|
|
import org.apache.hadoop.yarn.state.InvalidStateTransitonException;
|
|
|
import org.apache.hadoop.yarn.state.SingleArcTransition;
|
|
|
import org.apache.hadoop.yarn.state.StateMachine;
|
|
@@ -132,7 +132,8 @@ public abstract class RMStateStore extends AbstractService {
|
|
|
LOG.info("Storing info for app: " + appId);
|
|
|
try {
|
|
|
store.storeApplicationStateInternal(appId, appStateData);
|
|
|
- store.notifyDoneStoringApplication(appId, null);
|
|
|
+ store.notifyApplication(new RMAppEvent(appId,
|
|
|
+ RMAppEventType.APP_NEW_SAVED));
|
|
|
} catch (Exception e) {
|
|
|
LOG.error("Error storing app: " + appId, e);
|
|
|
store.notifyStoreOperationFailed(e);
|
|
@@ -156,7 +157,8 @@ public abstract class RMStateStore extends AbstractService {
|
|
|
LOG.info("Updating info for app: " + appId);
|
|
|
try {
|
|
|
store.updateApplicationStateInternal(appId, appStateData);
|
|
|
- store.notifyDoneUpdatingApplication(appId, null);
|
|
|
+ store.notifyApplication(new RMAppEvent(appId,
|
|
|
+ RMAppEventType.APP_UPDATE_SAVED));
|
|
|
} catch (Exception e) {
|
|
|
LOG.error("Error updating app: " + appId, e);
|
|
|
store.notifyStoreOperationFailed(e);
|
|
@@ -205,8 +207,9 @@ public abstract class RMStateStore extends AbstractService {
|
|
|
}
|
|
|
store.storeApplicationAttemptStateInternal(attemptState.getAttemptId(),
|
|
|
attemptStateData);
|
|
|
- store.notifyDoneStoringApplicationAttempt(attemptState.getAttemptId(),
|
|
|
- null);
|
|
|
+ store.notifyApplicationAttempt(new RMAppAttemptEvent
|
|
|
+ (attemptState.getAttemptId(),
|
|
|
+ RMAppAttemptEventType.ATTEMPT_NEW_SAVED));
|
|
|
} catch (Exception e) {
|
|
|
LOG.error("Error storing appAttempt: " + attemptState.getAttemptId(), e);
|
|
|
store.notifyStoreOperationFailed(e);
|
|
@@ -233,8 +236,9 @@ public abstract class RMStateStore extends AbstractService {
|
|
|
}
|
|
|
store.updateApplicationAttemptStateInternal(attemptState.getAttemptId(),
|
|
|
attemptStateData);
|
|
|
- store.notifyDoneUpdatingApplicationAttempt(attemptState.getAttemptId(),
|
|
|
- null);
|
|
|
+ store.notifyApplicationAttempt(new RMAppAttemptEvent
|
|
|
+ (attemptState.getAttemptId(),
|
|
|
+ RMAppAttemptEventType.ATTEMPT_UPDATE_SAVED));
|
|
|
} catch (Exception e) {
|
|
|
LOG.error("Error updating appAttempt: " + attemptState.getAttemptId(), e);
|
|
|
store.notifyStoreOperationFailed(e);
|
|
@@ -801,47 +805,28 @@ public abstract class RMStateStore extends AbstractService {
|
|
|
}
|
|
|
rmDispatcher.getEventHandler().handle(new RMFatalEvent(type, failureCause));
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
@SuppressWarnings("unchecked")
|
|
|
/**
|
|
|
- * In (@link handleStoreEvent}, this method is called to notify the
|
|
|
- * application that new application is stored in state store
|
|
|
- * @param appId id of the application that has been saved
|
|
|
- * @param storedException the exception that is thrown when storing the
|
|
|
- * application
|
|
|
+ * This method is called to notify the application that
|
|
|
+ * new application is stored or updated in state store
|
|
|
+ * @param event App event containing the app id and event type
|
|
|
*/
|
|
|
- private void notifyDoneStoringApplication(ApplicationId appId,
|
|
|
- Exception storedException) {
|
|
|
- rmDispatcher.getEventHandler().handle(
|
|
|
- new RMAppNewSavedEvent(appId, storedException));
|
|
|
- }
|
|
|
-
|
|
|
- @SuppressWarnings("unchecked")
|
|
|
- private void notifyDoneUpdatingApplication(ApplicationId appId,
|
|
|
- Exception storedException) {
|
|
|
- rmDispatcher.getEventHandler().handle(
|
|
|
- new RMAppUpdateSavedEvent(appId, storedException));
|
|
|
+ private void notifyApplication(RMAppEvent event) {
|
|
|
+ rmDispatcher.getEventHandler().handle(event);
|
|
|
}
|
|
|
-
|
|
|
+
|
|
|
@SuppressWarnings("unchecked")
|
|
|
/**
|
|
|
- * In (@link handleStoreEvent}, this method is called to notify the
|
|
|
- * application attempt that new attempt is stored in state store
|
|
|
- * @param appAttempt attempt that has been saved
|
|
|
+ * This method is called to notify the application attempt
|
|
|
+ * that new attempt is stored or updated in state store
|
|
|
+ * @param event App attempt event containing the app attempt
|
|
|
+ * id and event type
|
|
|
*/
|
|
|
- private void notifyDoneStoringApplicationAttempt(ApplicationAttemptId attemptId,
|
|
|
- Exception storedException) {
|
|
|
- rmDispatcher.getEventHandler().handle(
|
|
|
- new RMAppAttemptNewSavedEvent(attemptId, storedException));
|
|
|
+ private void notifyApplicationAttempt(RMAppAttemptEvent event) {
|
|
|
+ rmDispatcher.getEventHandler().handle(event);
|
|
|
}
|
|
|
-
|
|
|
- @SuppressWarnings("unchecked")
|
|
|
- private void notifyDoneUpdatingApplicationAttempt(ApplicationAttemptId attemptId,
|
|
|
- Exception updatedException) {
|
|
|
- rmDispatcher.getEventHandler().handle(
|
|
|
- new RMAppAttemptUpdateSavedEvent(attemptId, updatedException));
|
|
|
- }
|
|
|
-
|
|
|
+
|
|
|
/**
|
|
|
* EventHandler implementation which forward events to the FSRMStateStore
|
|
|
* This hides the EventHandle methods of the store from its public interface
|