Flink: Disable classloader check in TestIcebergSourceWithWatermarkExtractor to fix flakiness

[manuzhang]

I've seen the following flakiness several times. This PR fixes it by disabling Flink classloader check.

TestIcebergSourceWithWatermarkExtractor > testThrottling FAILED
    java.lang.AssertionError: 
    Expecting
      <CompletableFuture[Failed with the following stack trace:
    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.util.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.ensureInner(FlinkUserCodeClassLoaders.java:184)
    	at org.apache.flink.util.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:192)
    	at java.lang.Class.forName0(Native Method)
    	at java.lang.Class.forName(Class.java:348)
    	at com.esotericsoftware.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:136)
    	at com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:115)
    	at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
    	at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)
    	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:400)
    	at org.apache.flink.streaming.api.operators.collect.CollectCoordinationResponse.getResults(CollectCoordinationResponse.java:84)
    	at org.apache.flink.streaming.api.operators.collect.AbstractCollectResultBuffer.addResults(AbstractCollectResultBuffer.java:127)
    	at org.apache.flink.streaming.api.operators.collect.AbstractCollectResultBuffer.dealWithResponse(AbstractCollectResultBuffer.java:91)
    	at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.next(CollectResultFetcher.java:147)
    	at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:106)
    	at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.hasNext(CollectResultIterator.java:80)
    	at org.apache.iceberg.flink.source.TestIcebergSourceWithWatermarkExtractor.lambda$waitForRecords$6(TestIcebergSourceWithWatermarkExtractor.java:415)
    	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604)
    	at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1596)
    	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    	at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    	at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    	at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:175)
    ]>
    to be completed within 2M.

[pvary]

Thanks @manuzhang for the fix.
CC: @stevenzwu

[ajantha-bhat]

I am using the latest code (double checked) and the test case is still flaky.
https://github.com/apache/iceberg/actions/runs/7438225898/job/20236774791?pr=9298

TestIcebergSourceWithWatermarkExtractor > testThrottling FAILED
    java.lang.AssertionError: 
    Expecting
      <CompletableFuture[Failed with the following stack trace:
    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.util.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.ensureInner(FlinkUserCodeClassLoaders.java:184)
    	at org.apache.flink.util.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:192)
    	at java.lang.Class.forName0(Native Method)
    	at java.lang.Class.forName(Class.java:348)
    	at com.esotericsoftware.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:136)
    	at com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:115)
    	at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:641)
    	at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:752)
    	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:402)

I propose to disable this testcase (and run local fork to figure out the problem) to avoid causing problems for PR builders.
cc: @pvary, @manuzhang, @stevenzwu, @nastra, @Fokko

[manuzhang]

@ajantha-bhat The error looks weird. There's no way SafetyNetWrapperClassLoader is still used when classloader check has been disabled

[pvary]

Anybody can reproduce this issue locally?
I was able to run 1000 runs using IntelliJ without an issue, but I might miss something

[pvary]

Could not repro the case, but found the issue again in some tests, so decide to remove the checks for the record from the test (relying only on the metrics to check if the WA is working). This is the same approach as it is done in the Flink code, so it should be ok too. #9451