[ https://issues.apache.org/jira/browse/FLINK-34402?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17815132#comment-17815132 ]
yisha zhou edited comment on FLINK-34402 at 2/7/24 6:58 AM: ------------------------------------------------------------ Hi [~xtsong] , For the first question, yes, it's an ITCase that I'm going to add. The code is only for internal use in our company and not intended to be contributed to the community. But I can provide the keys to reproduce the issue. It's ITcase for Redis SQL Connector. I added a RunWith and PrepareForTest annotation, and then created a StreamTableEnvironment and used it to execute queries. Those queries would make mocked class to work and be checked. {code:java} @RunWith(PowerMockRunner.class) @PrepareForTest(RedisClientTableUtils.class) ··· EnvironmentSettings envSettings = EnvironmentSettings.newInstance().inStreamingMode().build(); tEnv = StreamTableEnvironment.create(env, envSettings);{code} Then an exception is thrown: {code:java} Caused by: java.lang.RuntimeException: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster. at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:321) at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:75) at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602) at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577) at java.util.concurrent.CompletableFuture$Completion.exec(CompletableFuture.java:443) 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:157) Caused by: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster. at org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess.lambda$new$0(DefaultJobMasterServiceProcess.java:97) at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760) at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736) at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1595) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: java.util.concurrent.CompletionException: java.lang.ClassCastException: org.apache.flink.api.common.ExecutionConfig cannot be cast to org.apache.flink.api.common.ExecutionConfig at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273) at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280) at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592) ... 3 more Caused by: java.lang.ClassCastException: org.apache.flink.api.common.ExecutionConfig cannot be cast to org.apache.flink.api.common.ExecutionConfig at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:98) at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:119) at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:403) at org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:379) at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:123) at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:95) at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112) at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590) ... 3 more {code} For question 2, I found if no users jars exist the system classLoader will be used in [BlobLibraryCacheManager#242|https://github.com/apache/flink/blob/f1fba33d85a802b896170ff3cdb0107ee082c44a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java#L242] . The related issue is FLINK-32265 The fixed code should be: {code:java} private UserCodeClassLoader getOrResolveClassLoader( Collection<PermanentBlobKey> libraries, Collection<URL> classPaths) throws IOException { synchronized (lockObject) { verifyIsNotReleased(); if (resolvedClassLoader == null) { boolean systemClassLoader = wrapsSystemClassLoader && libraries.isEmpty() && classPaths.isEmpty(); resolvedClassLoader = new ResolvedClassLoader( systemClassLoader ? Thread.currentThread().getContextClassLoader() : createUserCodeClassLoader( jobId, libraries, classPaths), libraries, classPaths, systemClassLoader); } else { resolvedClassLoader.verifyClassLoader(libraries, classPaths); } return resolvedClassLoader; } }{code} And thanks for your advice, for our internal use, we depend on powerMock to check calling times for all APIs and the output RowDatas for the queries. Do you get any advice for the scenario? was (Author: nilerzhou): Hi [~xtsong] , For the first question, yes, it's an ITCase that I'm going to add. The code is only for internal use in our company and not intended to be contributed to the community. But I can provide the keys to reproduce the issue. It's ITcase for Redis SQL Connector. I added a RunWith and PrepareForTest annotation, and then created a StreamTableEnvironment and used it to execute queries. Those queries would make mocked class to work and be checked. {code:java} @RunWith(PowerMockRunner.class) @PrepareForTest(RedisClientTableUtils.class) ··· EnvironmentSettings envSettings = EnvironmentSettings.newInstance().inStreamingMode().build(); tEnv = StreamTableEnvironment.create(env, envSettings);{code} Then an exception is thrown: {code:java} Caused by: java.lang.RuntimeException: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster. at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:321) at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:75) at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602) at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577) at java.util.concurrent.CompletableFuture$Completion.exec(CompletableFuture.java:443) 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:157) Caused by: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster. at org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess.lambda$new$0(DefaultJobMasterServiceProcess.java:97) at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760) at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736) at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1595) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: java.util.concurrent.CompletionException: java.lang.ClassCastException: org.apache.flink.api.common.ExecutionConfig cannot be cast to org.apache.flink.api.common.ExecutionConfig at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273) at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280) at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592) ... 3 more Caused by: java.lang.ClassCastException: org.apache.flink.api.common.ExecutionConfig cannot be cast to org.apache.flink.api.common.ExecutionConfig at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:98) at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:119) at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:403) at org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:379) at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:123) at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:95) at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112) at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590) ... 3 more {code} For question 2, I found if no users jars exist the system classLoader will be used in [BlobLibraryCacheManager#242|https://github.com/apache/flink/blob/f1fba33d85a802b896170ff3cdb0107ee082c44a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java#L242] . The related issue is [FLINK-32265|https://issues.apache.org/jira/browse/FLINK-32265] The fixed code should be: {code:java} private UserCodeClassLoader getOrResolveClassLoader( Collection<PermanentBlobKey> libraries, Collection<URL> classPaths) throws IOException { synchronized (lockObject) { verifyIsNotReleased(); if (resolvedClassLoader == null) { boolean systemClassLoader = wrapsSystemClassLoader && libraries.isEmpty() && classPaths.isEmpty(); resolvedClassLoader = new ResolvedClassLoader( systemClassLoader ? Thread.currentThread().getContextClassLoader() : createUserCodeClassLoader( jobId, libraries, classPaths), libraries, classPaths, systemClassLoader); } else { resolvedClassLoader.verifyClassLoader(libraries, classPaths); } return resolvedClassLoader; } }{code} And thanks for your advice, for our internal use, we depend on powerMock to check calling times for all APIs and the output RowDatas for the queries. Do you get any advice for the scenario? > Class loading conflicts when using PowerMock in ITcase. > -------------------------------------------------------- > > Key: FLINK-34402 > URL: https://issues.apache.org/jira/browse/FLINK-34402 > Project: Flink > Issue Type: Bug > Components: Runtime / Coordination > Affects Versions: 1.19.0 > Reporter: yisha zhou > Assignee: yisha zhou > Priority: Major > Fix For: 1.19.0 > > > Currently when no user jars exist, system classLoader will be used to load > classes as default. However, if we use powerMock to create some ITCases, the > framework will utilize JavassistMockClassLoader to load classes. Forcing the > use of the system classLoader can lead to class loading conflict issue. > Therefore we should use Thread.currentThread().getContextClassLoader() > instead of > ClassLoader.getSystemClassLoader() here. -- This message was sent by Atlassian Jira (v8.20.10#820010)