目前採用的Flink的版本是1.4.2，執行在yarn上，總是時不時的報錯“Invalid AMRMToken from appattempt”，導致AM掛掉。

簡而言之，就是AM和RM溝通的過程中，突然AM提供的Token不被認可，導致拒絕連線，進而AM掛掉。

後來發現早期版本的yarn-client是存在問題的，至少在2.6.0還是存在問題，在2.7.0之後解決了。

具體的錯誤描述可參見：https://issues.apache.org/jira/browse/YARN-3103。

但由於Flink為了減少依賴，將相關的依賴直接轉換成自己的類，去看了下這部分程式碼，果然是有問題。

更新了flink-shaded-hadoop專案中flink-shaded-hadoop2-uber專案引用的hadoop的包的版本，原先引用的是2.4.0，改完2.7.2，重新打包，部署。

正確的程式碼是：

private void updateAMRMToken(org.apache.hadoop.yarn.api.records.Token token)
    throws IOException
  {
    org.apache.hadoop.security.token.Token<AMRMTokenIdentifier> amrmToken = new
 
 org.apache.hadoop.security.token.Token(token.getIdentifier().array(), token.getPassword().array(), new Text(token.getKind()), new Text(token.getService()));
    
    UserGroupInformation currentUGI = UserGroupInformation.getCurrentUser();
    currentUGI.addToken(amrmToken);
    amrmToken.setService(ClientRMProxy.getAMRMTokenService(getConfig()));
  }
 

在構建完了token之後，才會更新服務。

錯誤的程式碼是：

private void updateAMRMToken(org.apache.hadoop.yarn.api.records.Token token)
    throws IOException
  {
    org.apache.hadoop.security.token.Token<AMRMTokenIdentifier> amrmToken = new org.apache.hadoop.security.token.Token(token.getIdentifier().array(), token.getPassword().array(), new Text(token.getKind()), new Text(token.getService()));
    
    amrmToken.setService(ClientRMProxy.getAMRMTokenService(getConfig()));
    UserGroupInformation currentUGI = UserGroupInformation.getCurrentUser();
    if (UserGroupInformation.isSecurityEnabled()) {
      currentUGI = UserGroupInformation.getLoginUser();
    }
    currentUGI.addToken(amrmToken);
  }

在增加token之前，先設定了service，導致上下文發生變化，就導致了多個不同的token的產生，後續如果選擇了特定的token，就會報錯。