diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java index 7783662..ca61f4e 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java @@ -139,8 +139,8 @@ protected Version getCurrentVersion() { @Override protected synchronized Version loadVersion() throws Exception { Path versionNodePath = getNodePath(rootDirPath, VERSION_NODE); - if (fs.exists(versionNodePath)) { - FileStatus status = fs.getFileStatus(versionNodePath); + FileStatus status = getFileStatus(versionNodePath); + if (status != null) { byte[] data = readFile(versionNodePath, status.getLen()); Version version = new VersionPBImpl(VersionProto.parseFrom(data)); @@ -165,9 +165,9 @@ protected synchronized void storeVersion() throws Exception { public synchronized long getAndIncrementEpoch() throws Exception { Path epochNodePath = getNodePath(rootDirPath, EPOCH_NODE); long currentEpoch = 0; - if (fs.exists(epochNodePath)) { + FileStatus status = getFileStatus(epochNodePath); + if (status != null) { // load current epoch - FileStatus status = fs.getFileStatus(epochNodePath); byte[] data = readFile(epochNodePath, status.getLen()); Epoch epoch = new EpochPBImpl(EpochProto.parseFrom(data)); currentEpoch = epoch.getEpoch(); @@ -466,7 +466,7 @@ public synchronized void removeRMDelegationTokenState( } @Override - protected void updateRMDelegationTokenState( + protected synchronized void updateRMDelegationTokenState( RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate) throws Exception { storeOrUpdateRMDelegationTokenState(rmDTIdentifier, renewDate, true); @@ -560,6 +560,14 @@ private void deleteFile(Path deletePath) throws Exception { } } + private FileStatus getFileStatus(Path path) throws Exception { + try { + return fs.getFileStatus(path); + } catch (FileNotFoundException e) { + return null; + } + } + /* * In order to make this write atomic as a part of write we will first write * data to .tmp file and then rename it. Here we are assuming that rename is diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java index 3646949..8cd776e 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java @@ -91,15 +91,16 @@ protected synchronized void closeInternal() throws Exception { } @Override - public void storeApplicationStateInternal( + public synchronized void storeApplicationStateInternal( ApplicationId appId, ApplicationStateData appState) throws Exception { state.appState.put(appId, appState); } @Override - public void updateApplicationStateInternal(ApplicationId appId, - ApplicationStateData appState) throws Exception { + public synchronized void updateApplicationStateInternal( + ApplicationId appId, ApplicationStateData appState) + throws Exception { LOG.info("Updating final state " + appState.getState() + " for app: " + appId); if (state.appState.get(appId) != null) { @@ -186,7 +187,7 @@ public synchronized void removeRMDelegationTokenState( } @Override - protected void updateRMDelegationTokenState( + protected synchronized void updateRMDelegationTokenState( RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate) throws Exception { removeRMDelegationTokenState(rmDTIdentifier); @@ -237,7 +238,7 @@ protected Version getCurrentVersion() { } @Override - public void storeOrUpdateAMRMTokenSecretManagerState( + public synchronized void storeOrUpdateAMRMTokenSecretManagerState( AMRMTokenSecretManagerState amrmTokenSecretManagerState, boolean isUpdate) { if (amrmTokenSecretManagerState != null) {