[FLINK-19843][table] ParquetFsStreamingSinkITCase.testPart failed with 'Trying to access closed classloader'

[JingsongLi]

What is the purpose of the change

ParquetFsStreamingSinkITCase.testPart failed with 'Trying to access closed classloader'

Instance: https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=8714&view=logs&j=ba53eb01-1462-56a3-8e98-0dd97fbcaab5&t=bfbc6239-57a0-5db0-63f3-41551b4f7d51

Brief change log

• ParquetFsStreamingSinkITCase should not create a new execution env in local model. That would lead to confusion.
• We should close iterator for TableResult.collect

Verifying this change

in flink-parquet: mvn -Dflink.forkCount=2 -Dflink.forkCountTestPackage=2 -Dfast -Dinclude_hadoop_aws -Dhadoop.version=3.1.3 install

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive):no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no
• The S3 file system connector: (no

Documentation

• Does this pull request introduce a new feature? (no

[flinkbot]

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 3d62293 (Mon Nov 02 06:36:55 UTC 2020)

Warnings:

• No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

• ❓ 1. The [description] looks good.
• ❓ 2. There is [consensus] that the contribution should go into to Flink.
• ❓ 3. Needs [attention] from.
• ❓ 4. The change fits into the overall [architecture].
• ❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
• @flinkbot approve all to approve all aspects
• @flinkbot approve-until architecture to approve everything until architecture
• @flinkbot attention @username1 [@username2 ..] to require somebody's attention
• @flinkbot disapprove architecture to remove an approval you gave earlier

[flinkbot]

CI report:

• 3d62293 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run travis re-run the last Travis build
• @flinkbot run azure re-run the last Azure build

[dianfu]

@JingsongLi Thanks a lot for the quick fix!

I have verified that the tests could pass with this PR. However, there are output as following which seems a little wired. Could you take a look if these logs are as expected?

[INFO] Running org.apache.flink.formats.parquet.ParquetFsStreamingSinkITCase
[INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 13.597 s - in org.apache.flink.formats.parquet.ParquetFsStreamingSinkITCase
Exception in thread "Thread-10" 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:161)
	at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.getResource(FlinkUserCodeClassLoaders.java:179)
	at org.apache.hadoop.conf.Configuration.getResource(Configuration.java:2780)
	at org.apache.hadoop.conf.Configuration.getStreamReader(Configuration.java:3036)
	at org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2995)
	at org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2968)
	at org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2848)
	at org.apache.hadoop.conf.Configuration.get(Configuration.java:1200)
	at org.apache.hadoop.conf.Configuration.getTimeDuration(Configuration.java:1812)
	at org.apache.hadoop.conf.Configuration.getTimeDuration(Configuration.java:1789)
	at org.apache.hadoop.util.ShutdownHookManager.getShutdownTimeout(ShutdownHookManager.java:183)
	at org.apache.hadoop.util.ShutdownHookManager.shutdownExecutor(ShutdownHookManager.java:145)
	at org.apache.hadoop.util.ShutdownHookManager.access$300(ShutdownHookManager.java:65)
	at org.apache.hadoop.util.ShutdownHookManager$1.run(ShutdownHookManager.java:102)
[INFO] Running org.apache.flink.formats.parquet.ParquetTableSourceITCase
[INFO] Tests run: 22, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 28.703 s - in org.apache.flink.formats.parquet.ParquetFileSystemITCase
Exception in thread "Thread-14" 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:161)
	at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.getResource(FlinkUserCodeClassLoaders.java:179)
	at org.apache.hadoop.conf.Configuration.getResource(Configuration.java:2780)
	at org.apache.hadoop.conf.Configuration.getStreamReader(Configuration.java:3036)
	at org.apache.hadoop.conf.Configuration.loadResource(Configuration.java:2995)
	at org.apache.hadoop.conf.Configuration.loadResources(Configuration.java:2968)
	at org.apache.hadoop.conf.Configuration.getProps(Configuration.java:2848)
	at org.apache.hadoop.conf.Configuration.get(Configuration.java:1200)
	at org.apache.hadoop.conf.Configuration.getTimeDuration(Configuration.java:1812)
	at org.apache.hadoop.conf.Configuration.getTimeDuration(Configuration.java:1789)
	at org.apache.hadoop.util.ShutdownHookManager.getShutdownTimeout(ShutdownHookManager.java:183)
	at org.apache.hadoop.util.ShutdownHookManager.shutdownExecutor(ShutdownHookManager.java:145)
	at org.apache.hadoop.util.ShutdownHookManager.access$300(ShutdownHookManager.java:65)
	at org.apache.hadoop.util.ShutdownHookManager$1.run(ShutdownHookManager.java:102)
[INFO] Running org.apache.flink.formats.parquet.protobuf.ParquetProtoStreamingFileSinkITCase
[INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 8.394 s - in org.apache.flink.formats.parquet.ParquetTableSourceITCase
[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 4.714 s - in org.apache.flink.formats.parquet.protobuf.ParquetProtoStreamingFileSinkITCase
[INFO] 
[INFO] Results:
[INFO] 
[INFO] Tests run: 31, Failures: 0, Errors: 0, Skipped: 0```

[JingsongLi]

Hi @dianfu , Thanks for the verification!
This is because Hadoop 3 starts asynchronous threads to execute some shutdown hooks.
These hooks are run after the job is executed, as a result, the classloader has been released, but in hooks, configuration still holds the released classloader, so it will fail to throw an exception in this asynchronous thread.

I can't find a proper way to avoid it (exception logs), and if it doesn't affect our testing, I'll consider ignoring it directly.

[dianfu]

@JingsongLi Thanks a lot for the explanation! I think we need to figure out if it has side effects for normal jobs. If so, it will be a problem which needs to be addressed. However, it seems that it's not related to this PR. Could you create a separate JIRA to track this issue?

[JingsongLi]

I created https://issues.apache.org/jira/browse/FLINK-19916 for track it.

[dianfu]

@JingsongLi Thanks a lot! This PR LGTM. +1 from my side.