Flink on yarn的問題：Invalid AMRMToken

目前采用的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，就會報錯。