[SPARK-27301][Dstream]Shorten the FileSystem cached life cycle to the cleanup method inner scope

[yaooqinn]

What changes were proposed in this pull request?

The cached FileSystem's token will expire if no tokens explicitly are add into it.

19/03/28 13:40:16 INFO storage.BlockManager: Removing RDD 83189
19/03/28 13:40:16 INFO rdd.MapPartitionsRDD: Removing RDD 82860 from persistence list
19/03/28 13:40:16 INFO spark.ContextCleaner: Cleaned shuffle 6005
19/03/28 13:40:16 INFO storage.BlockManager: Removing RDD 82860
19/03/28 13:40:16 INFO scheduler.ReceivedBlockTracker: Deleting batches:
19/03/28 13:40:16 INFO scheduler.InputInfoTracker: remove old batch metadata: 1553750250000 ms
19/03/28 13:40:17 WARN security.UserGroupInformation: PriviledgedActionException as:urs@HADOOP.HZ.NETEASE.COM (auth:KERBEROS) cause:org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.token.SecretManager$InvalidToken): token (HDFS_DELEGATION_TOKEN token 53240500 for urs) is expired, current time: 2019-03-28 13:40:17,010+0800 expected renewal time: 2019-03-28 13:39:48,523+0800
19/03/28 13:40:17 WARN ipc.Client: Exception encountered while connecting to the server : org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.token.SecretManager$InvalidToken): token (HDFS_DELEGATION_TOKEN token 53240500 for urs) is expired, current time: 2019-03-28 13:40:17,010+0800 expected renewal time: 2019-03-28 13:39:48,523+0800
19/03/28 13:40:17 WARN security.UserGroupInformation: PriviledgedActionException as:urs@HADOOP.HZ.NETEASE.COM (auth:KERBEROS) cause:org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.token.SecretManager$InvalidToken): token (HDFS_DELEGATION_TOKEN token 53240500 for urs) is expired, current time: 2019-03-28 13:40:17,010+0800 expected renewal time: 2019-03-28 13:39:48,523+0800
19/03/28 13:40:17 WARN hdfs.LeaseRenewer: Failed to renew lease for [DFSClient_NONMAPREDUCE_-1396157959_1] for 53 seconds. Will retry shortly ...
org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.token.SecretManager$InvalidToken): token (HDFS_DELEGATION_TOKEN token 53240500 for urs) is expired, current time: 2019-03-28 13:40:17,010+0800 expected renewal time: 2019-03-28 13:39:48,523+0800
at org.apache.hadoop.ipc.Client.call(Client.java:1468)
at org.apache.hadoop.ipc.Client.call(Client.java:1399)
at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:232)
at com.sun.proxy.$Proxy11.renewLease(Unknown Source)
at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.renewLease(ClientNamenodeProtocolTranslatorPB.java:571)
at sun.reflect.GeneratedMethodAccessor40.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:187)
at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
at com.sun.proxy.$Proxy12.renewLease(Unknown Source)
at org.apache.hadoop.hdfs.DFSClient.renewLease(DFSClient.java:878)
at org.apache.hadoop.hdfs.LeaseRenewer.renew(LeaseRenewer.java:417)
at org.apache.hadoop.hdfs.LeaseRenewer.run(LeaseRenewer.java:442)
at org.apache.hadoop.hdfs.LeaseRenewer.access$700(LeaseRenewer.java:71)
at org.apache.hadoop.hdfs.LeaseRenewer$1.run(LeaseRenewer.java:298)
at java.lang.Thread.run(Thread.java:748)

This PR shorten the FileSystem cached life cycle to the cleanup method inner scope in case of token expiry.

How was this patch tested?

existing ut

[SparkQA]

Test build #104042 has finished for PR 24235 at commit 8d87a3f.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[gaborgsomogyi]

Maybe it's better approach to re-create it when it throws exception.

[gaborgsomogyi]

Just for the sake of my understanding when will be the filesystem re-created with the

• old
• new

implementation (how much is the gain)?

[attilapiros]

Interesting. The path.hetFileSystem(...) already gives back a cached FileSystem if caching for that schema is explicitly not disabled, see org.apache.hadoop.fs.FileSystem#get(java.net.URI, org.apache.hadoop.conf.Configuration)

      String disableCacheName = String.format("fs.%s.impl.disable.cache", scheme);
      return conf.getBoolean(disableCacheName, false) ? createFileSystem(uri, conf) : CACHE.get(uri, conf);

Is the caching disabled?
If not my question would be what happens when the token for that cached filesystem is expired?
I assume you get the original error from a long running test, have you tested this change as well with the same long running test?

[yaooqinn]

token expiration has no exception but only warning messages @gaborgsomogyi

[yaooqinn]

@attilapiros the word “cached” you mentioned is how to get or create a fs instance，here “cached” means we hold an once created fs instance as a class member

[gaborgsomogyi]

time: 2019-03-28 13:40:17,010+0800 expected renewal time: 2019-03-28 13:39:48,523+0800

and

19/03/28 13:40:17 WARN hdfs.LeaseRenewer: Failed to renew lease for [DFSClient_NONMAPREDUCE_-1396157959_1] for 53 seconds. Will retry shortly ...

Why do you think a new instance will do things better?

[attilapiros]

What I wanted to point out is if you get an already cached instance from FileSystem#get where the renewal of token is also missing you run into the same problem.

I would like to understand the problem well before I would say this solution is fine.

So could you please answer all my questions? Most importantly:

I assume you get the original error from a long running test, have you tested this change as well with the same long running test?

[attilapiros]

The reason why I ask this as you only mentioned:

existing ut

at the testing section of the PR description.

[yaooqinn]

@attilapiros thanks for point out what your real question is. I will do a manual test. BTW，cache is disabled explicitly in my case.

[srowen]

I think this change is fine, in that it at least pushes the caching issue down into the library. No particular reason to 'cache' in Spark. However i agree it may not resolve the problem unless that library caching is disabled.

[attilapiros]

Yes.
So @yaooqinn I would like to kindly ask you to run another experiment without fs.<scheme>.impl.disable.cache too for checking whether the correction is complete.

[srowen]

(Even if it doesn't I don't think there's any harm in this change, as indeed the fact that Hadoop caches things makes this construction kind of irrelevant)

[gaborgsomogyi]

I think this change is fine

Delegation tokens are obtained for 24 hours and renewed after 24h * 0.75 per default for safety reasons. So asking again just like before: Without SecurityManager obtaining a new token how does a new instance solve this issue?

[gaborgsomogyi]

I don't think there's any harm in this change

I agree.

[yaooqinn]

@srowen Yes, this change only covers the issues when fs cache disabled, not fixes the issues belong to underlying libraries, such as https://issues.apache.org/jira/browse/HDFS-9276

[yaooqinn]

If without cached disabled, there will be various places to modify to deal with underlying file system issues which might only exists in some particular lower version hadoops,
and we may do getFileSystem each time like this

val fileSystem = path.getFileSystem(hadoopConf)

fileSystem.addDelegationTokens(UserGroupInformation.getCurrentUser.getCredentials.getAllTokens)

[yaooqinn]

The other thing to note is that the log in the pr description is where the warning message first appeared，and it did not stop the main streaming schedule logic and the application kept on running.

[attilapiros]

@yaooqinn just to have the same understanding:
From your previous comments I assume your manual test failed even beside disabled caching (fs.<scheme>.impl.disable.cache=true).

As the delete in line 90 is in try-catch block where Exception is matched and there is no re-throw continuing with warning is very much expected:

spark/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStreamCheckpointData.scala

Lines 85 to 97 in 8d87a3f

	try {
	val path = new Path(file)
	if (fileSystem == null) {
	fileSystem = path.getFileSystem(dstream.ssc.sparkContext.hadoopConfiguration)
	}
	fileSystem.delete(path, true)
	timeToCheckpointFile -= time
	logInfo("Deleted checkpoint file '" + file + "' for time " + time)
	} catch {
	case e: Exception =>
	logWarning("Error deleting old checkpoint file '" + file + "' for time " + time, e)
	fileSystem = null
	}

[yaooqinn]

the application is belong to one of clients, I am not capable to run the same case

This is the last warning message I got at that moment

19/03/28 13:44:05 INFO scheduler.TaskSetManager: Finished task 21.0 in stage 17071.0 (TID 1817652) in 210 ms on hadoop1479.hz.163.org (executor 2) (22/96)
19/03/28 13:44:06 WARN security.UserGroupInformation: PriviledgedActionException as:urs@HADOOP.HZ.NETEASE.COM (auth:KERBEROS) cause:org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.token.SecretManager$InvalidToken): token (HDFS_DELEGATION_TOKEN token 53240500 for urs) is expired, current time: 2019-03-28 13:44:06,189+0800 expected renewal time: 2019-03-28 13:39:48,523+0800
19/03/28 13:44:06 WARN ipc.Client: Exception encountered while connecting to the server : org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.token.SecretManager$InvalidToken): token (HDFS_DELEGATION_TOKEN token 53240500 for urs) is expired, current time: 2019-03-28 13:44:06,189+0800 expected renewal time: 2019-03-28 13:39:48,523+0800
19/03/28 13:44:06 WARN security.UserGroupInformation: PriviledgedActionException as:urs@HADOOP.HZ.NETEASE.COM (auth:KERBEROS) cause:org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.token.SecretManager$InvalidToken): token (HDFS_DELEGATION_TOKEN token 53240500 for urs) is expired, current time: 2019-03-28 13:44:06,189+0800 expected renewal time: 2019-03-28 13:39:48,523+0800
19/03/28 13:44:06 WARN hdfs.LeaseRenewer: Failed to renew lease for [DFSClient_NONMAPREDUCE_-1396157959_1] for 282 seconds.  Will retry shortly ...
org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.token.SecretManager$InvalidToken): token (HDFS_DELEGATION_TOKEN token 53240500 for urs) is expired, current time: 2019-03-28 13:44:06,189+0800 expected renewal time: 2019-03-28 13:39:48,523+0800
	at org.apache.hadoop.ipc.Client.call(Client.java:1468)
	at org.apache.hadoop.ipc.Client.call(Client.java:1399)
	at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:232)
	at com.sun.proxy.$Proxy11.renewLease(Unknown Source)
	at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.renewLease(ClientNamenodeProtocolTranslatorPB.java:571)
	at sun.reflect.GeneratedMethodAccessor40.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:606)
	at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:187)
	at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
	at com.sun.proxy.$Proxy12.renewLease(Unknown Source)
	at org.apache.hadoop.hdfs.DFSClient.renewLease(DFSClient.java:878)
	at org.apache.hadoop.hdfs.LeaseRenewer.renew(LeaseRenewer.java:417)
	at org.apache.hadoop.hdfs.LeaseRenewer.run(LeaseRenewer.java:442)
	at org.apache.hadoop.hdfs.LeaseRenewer.access$700(LeaseRenewer.java:71)
	at org.apache.hadoop.hdfs.LeaseRenewer$1.run(LeaseRenewer.java:298)
	at java.lang.Thread.run(Thread.java:748)

[yaooqinn]

The user modified the app, removed the checkpoint operation and re-run, the app has running for 45 hours without those warnings and I think I need to wait at least one week to get the result.

[srowen]

Merged to master/2.4/2.3