[Feature Request]: FileBasedSink$Writer failed to close channel

[gabrywu]

What would you like to happen?

org.apache.hadoop.hdfs.DataStreamer throws a warning message: DataStreamer Exception java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration 'classloader.check-leaked-classloader'.
at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.ensureInner(FlinkUserCodeClassLoaders.java:164) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:172) ~[flink-dist_2.12-1.13.5.jar:1.13.5]
at java.lang.Class.forName0(Native Method) ~[?:1.8.0_362]
at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_362]
at org.apache.hadoop.conf.Configuration.getClassByNameOrNull(Configuration.java:2625) ~[flink-shaded-hadoop-2-uber-3.3.3.1.0.11-10.0.jar:3.3.3.1.0.11-10.0]
at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:2590) ~[flink-shaded-hadoop-2-uber-3.3.3.1.0.11-10.0.jar:3.3.3.1.0.11-10.0]
at org.apache.hadoop.crypto.CryptoCodec.getCodecClasses(CryptoCodec.java:126) ~[flink-shaded-hadoop-2-uber-3.3.3.1.0.11-10.0.jar:3.3.3.1.0.11-10.0]
at org.apache.hadoop.crypto.CryptoCodec.getInstance(CryptoCodec.java:61) ~[flink-shaded-hadoop-2-uber-3.3.3.1.0.11-10.0.jar:3.3.3.1.0.11-10.0]

And then org.apache.beam.sdk.io.FileBasedSink$Writer will failed to close channel, which throws errors

org.apache.beam.sdk.io.FileBasedSink$Writer [] - Closing channel for /path/.temp-beam/39e45b649fb492e2-cf6a-471d-932b-fc33858aa49d failed.
java.io.IOException: java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration 'classloader.check-leaked-classloader'.
at org.apache.hadoop.hdfs.ExceptionLastSeen.set(ExceptionLastSeen.java:45) ~[flink-shaded-hadoop-2-uber-3.3.3.1.0.11-10.0.jar:3.3.3.1.0.11-10.0]
at org.apache.hadoop.hdfs.DataStreamer.run(DataStreamer.java:823) ~[flink-shaded-hadoop-2-uber-3.3.3.1.0.11-10.0.jar:3.3.3.1.0.11-10.0]
Caused by: java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration 'classloader.check-leaked-classloader'.

Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner