HDFS_DELEGATION_TOKEN這個BUG在很多文章中都出現(xiàn)著锐极，講了很多原理笙僚，但是只給出了官方引用地扯，完全沒有給出如何解決溪烤，我們線上的業(yè)務(wù)就有著這樣的問題味咳，7天一到馬上出現(xiàn)這問題了，官方明明說這個bug修復(fù)了呀檬嘀，因為我們使用的版本是比較新的槽驶，理論上不會有這樣的問題才對，可是偏偏就有了鸳兽，沒辦法掂铐，只能硬上了，花了兩天的時間找到了解決這個問題的辦法揍异，下面會還原這個錯誤及給出解決方案全陨。
版本列表
測試用例配置
添加 hdfs-site.xml 配置
dfs.namenode.delegation.key.update-interval=60000 #1分鐘
dfs.namenode.delegation.token.max-lifetime=180000 #3分鐘
dfs.namenode.delegation.token.renew-interval=60000 #1分鐘
修改 /etc/krb5.conf ticket過期為1小時
...
ticket_lifetime = 1h
...
代碼內(nèi)有kerberos認(rèn)證
class App {
System.setProperty("java.security.krb5.conf", "/etc/krb5.conf")
System.setProperty("sun.security.krb5.debug", "false")
val hConf = new Configuration
hConf.addResource("hbase-site.xml")
UserGroupInformation.setConfiguration(hConf)
UserGroupInformation.loginUserFromKeytab("hbase-bd@EXAMPLE.COM", "/etc/security/keytabs/hbase.headless.keytab")
val sparkConf = new SparkConf()
// .setMaster("local[12]")
.setAppName("HDFS_DELEGATION_TOKEN")
val spark = SparkSession
.builder
.config(sparkConf)
.getOrCreate()
hConf.set("hbase.mapreduce.inputtable", "test_log")
def run(args: Array[String]): Unit = {
val sc = spark.sparkContext
import spark.implicits._
val userRDD: RDD[Log] = sc.newAPIHadoopRDD(
hConf,
classOf[TableInputFormat],
classOf[ImmutableBytesWritable],
classOf[Result]
).flatMap {
rdd => {
val map = HbaseUtil.result2Map(rdd._2)
val log = Log(
map.get("uid")
)
Array(log)
}
}
userRDD.toDS().cache().createTempView("log")
spark.sql(
"""select * from log""".stripMargin)
.show(false)
spark.catalog.dropTempView("log")
userRDD.unpersist()
}
}
case class Log(uid: String)
object App {
def main(args: Array[String]): Unit = {
val app = new App()
while (true) {
app.run(args)
TimeUnit.MINUTES.sleep(3)
}
}
}
測試百度跟谷歌中最最最多出現(xiàn)的解決方案
--conf spark.hadoop.fs.hdfs.impl.disable.cache=true
--conf mapreduce.job.complete.cancel.delegation.tokens=false
- 測試提交
spark-submit --master yarn \
--class com.dounine.hbase.App \
--executor-memory 1g \
--driver-memory 1g \
--keytab /etc/security/keytabs/hbase.headless.keytab \
--principal hbase-bd@EXAMPLE.COM \
build/libs/hdfs-token-1.0.0-SNAPSHOT-all.jar
- 測試提交
...
--conf spark.hadoop.fs.hdfs.impl.disable.cache=true \
--conf mapreduce.job.complete.cancel.delegation.tokens=false \
...
- 測試提交
...
--conf mapreduce.job.complete.cancel.delegation.tokens=false \
...
- 測試提交
...
--conf spark.hadoop.fs.hdfs.impl.disable.cache=true \
...
1，2衷掷，3辱姨，4 測試結(jié)果
時間觀察3分鐘 => 正常
時間觀察10分鐘 => 正常
時間觀察30分鐘 => 正常
時間觀察60分鐘 => 正常
時間觀察120分鐘 => 正常
測試結(jié)論 => 與1、2戚嗅、3雨涛、4 --conf 配置無關(guān)
好吧枢舶，我已經(jīng)懷疑人生、可能是我打開的方式不對
繼續(xù)測試
將認(rèn)證代碼放入run方法內(nèi)
def run(args: Array[String]): Unit = {
System.setProperty("java.security.krb5.conf", "/etc/krb5.conf")
System.setProperty("sun.security.krb5.debug", "false")
val hConf = new Configuration
hConf.addResource("hbase-site.xml")
UserGroupInformation.setConfiguration(hConf)
UserGroupInformation.loginUserFromKeytab("hbase-bd@EXAMPLE.COM", "/etc/security/keytabs/hbase.headless.keytab")
val sparkConf = new SparkConf()
// .setMaster("local[12]")
.setAppName("HDFS_DELEGATION_TOKEN")
val spark = SparkSession
.builder
.config(sparkConf)
.getOrCreate()
hConf.set("hbase.mapreduce.inputtable", "test_log")
....
時間觀察3分鐘 => 正常
時間觀察6分鐘 => 異常
18/12/29 16:50:31 ERROR AsyncEventQueue: Listener EventLoggingListener threw an exception
org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.token.SecretManager$InvalidToken): token (token for hbase: HDFS_DELEGATION_TOKEN owner
=hbase-bd@EXAMPLE.COM, renewer=yarn, realUser=, issueDate=1546072104965, maxDate=1546072704965, sequenceNumber=15985, masterKeyId=748) is expired, curr
ent time: 2018-12-29 16:32:29,829+0800 expected renewal time: 2018-12-29 16:31:24,965+0800
at org.apache.hadoop.ipc.Client.getRpcResponse(Client.java:1497)
at org.apache.hadoop.ipc.Client.call(Client.java:1443)
at org.apache.hadoop.ipc.Client.call(Client.java:1353)
at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:228)
at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:116)
at com.sun.proxy.$Proxy11.fsync(Unknown Source)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.fsync(ClientNamenodeProtocolTranslatorPB.java:980)
at sun.reflect.GeneratedMethodAccessor11.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:422)
at org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invokeMethod(RetryInvocationHandler.java:165)
at org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invoke(RetryInvocationHandler.java:157)
at org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invokeOnce(RetryInvocationHandler.java:95)
at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:359)
at com.sun.proxy.$Proxy12.fsync(Unknown Source)
...
問題發(fā)現(xiàn)
通過幾十遍不斷的調(diào)整位置替久、最終確認(rèn)的問題所在
還有Exector的問題
UserGroupInformation.setConfiguration(hConf)
UserGroupInformation.loginUserFromKeytab("hbase-bd@EXAMPLE.COM", "/etc/security/keytabs/hbase.headless.keytab")
是由于以上兩句kerberos認(rèn)證代碼導(dǎo)致的結(jié)果
跟下面的配置沖突了
--principal $principal --keytab $keytab
解決方案(一)
刪除掉下面代碼中的這兩句認(rèn)證即可,使用--principal $principal --keytab $keytab
UserGroupInformation.setConfiguration(hConf)
UserGroupInformation.loginUserFromKeytab("hbase-bd@EXAMPLE.COM", "/etc/security/keytabs/hbase.headless.keytab")
因為Spark的--principal --keytab會在令牌即將過期的時候幫我們重新續(xù)定凉泄，如果代碼里面加上之后，Spark會讀取到ApplicationMaster中用戶已經(jīng)認(rèn)證了蚯根，沒有過期是不會續(xù)定NodeManager中的Exector的后众。
如果是開發(fā)環(huán)境模式，可以加一個判斷使用以上兩句代碼颅拦，簡單粗暴
解決方案(二)
使用UserGroupInformation的進(jìn)程認(rèn)證方式
spark.sparkContext
.parallelize(0 to 1000)
.repartition(10)
.foreachPartition {
iter => {
val hConf = new Configuration
hConf.addResource("hbase-site.xml")
val ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI("hbase-bd@EXAMPLE.COM", "/etc/security/keytabs/hbase.headless.keytab")
ugi.doAs(new PrivilegedAction[Unit] {//在每個Partition認(rèn)證
override def run(): Unit = {
val logDir = new Path(args(0))
val fs = FileSystem.get(hConf)
if (!fs.exists(logDir)) throw new Exception(logDir.toUri.getPath + " director not exist.")
while (iter.hasNext) {
iter.next()
val logPaths = fs.listFiles(logDir, false)
TimeUnit.MILLISECONDS.sleep(10)
}
}
})
}
}
BUG 7 天后再次出現(xiàn)
上面推導(dǎo)還是有問題蒂誉，還有望知道BUG解決的小伙伴告知一下。
臨時解決方案
就是加大過期的時間
