[ https://issues.apache.org/jira/browse/FLINK-25937?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17489997#comment-17489997 ]
Gen Luo commented on FLINK-25937: --------------------------------- The reason has been identified. Parallelism of a transformation with default parallelism(-1) is set when transforming, using the default parallelism set in the environment. However, in SinkExpander#expand, the environment parallelism is set to -1 at the entrance, to verify if the parallelism of a expanded transformation is set. The environment parallelism will be restored when exiting the method, but at present the transform is called within this scope. If the parallelism of a sink is not set, the parallelism of the sink transformation and all transformations expanded from it will not be handled, so the JobGraph generated will have vertices with -1 parallelism, causing the assertion failure in AdaptiveScheduler. We can fix the bug by putting the restoring of the environment parallelism ahead of transforming the sink transformations. The pull request has been created, and has been verified with UpsertKafkaTableITCase. > SQL Client end-to-end test e2e fails on AZP > ------------------------------------------- > > Key: FLINK-25937 > URL: https://issues.apache.org/jira/browse/FLINK-25937 > Project: Flink > Issue Type: Bug > Components: API / Core, API / DataStream, Runtime / Coordination, > Table SQL / API > Affects Versions: 1.15.0 > Reporter: Till Rohrmann > Assignee: Gen Luo > Priority: Blocker > Labels: pull-request-available, test-stability > > The {{SQL Client end-to-end test}} e2e tests fails on AZP when using the > {{AdaptiveScheduler}} because the scheduler expects that the parallelism is > set for all vertices: > {code} > Feb 03 03:45:13 org.apache.flink.runtime.client.JobInitializationException: > Could not start the JobMaster. > Feb 03 03:45:13 at > org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess.lambda$new$0(DefaultJobMasterServiceProcess.java:97) > Feb 03 03:45:13 at > java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774) > Feb 03 03:45:13 at > java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750) > Feb 03 03:45:13 at > java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) > Feb 03 03:45:13 at > java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1609) > Feb 03 03:45:13 at > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) > Feb 03 03:45:13 at > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) > Feb 03 03:45:13 at java.lang.Thread.run(Thread.java:748) > Feb 03 03:45:13 Caused by: java.util.concurrent.CompletionException: > java.lang.IllegalStateException: The adaptive scheduler expects the > parallelism being set for each JobVertex (violated JobVertex: > f74b775b58627a33e46b8c155b320255). > Feb 03 03:45:13 at > java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273) > Feb 03 03:45:13 at > java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280) > Feb 03 03:45:13 at > java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1606) > Feb 03 03:45:13 ... 3 more > Feb 03 03:45:13 Caused by: java.lang.IllegalStateException: The adaptive > scheduler expects the parallelism being set for each JobVertex (violated > JobVertex: f74b775b58627a33e46b8c155b320255). > Feb 03 03:45:13 at > org.apache.flink.util.Preconditions.checkState(Preconditions.java:215) > Feb 03 03:45:13 at > org.apache.flink.runtime.scheduler.adaptive.AdaptiveScheduler.assertPreconditions(AdaptiveScheduler.java:296) > Feb 03 03:45:13 at > org.apache.flink.runtime.scheduler.adaptive.AdaptiveScheduler.<init>(AdaptiveScheduler.java:230) > Feb 03 03:45:13 at > org.apache.flink.runtime.scheduler.adaptive.AdaptiveSchedulerFactory.createInstance(AdaptiveSchedulerFactory.java:122) > Feb 03 03:45:13 at > org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:115) > Feb 03 03:45:13 at > org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:345) > Feb 03 03:45:13 at > org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:322) > Feb 03 03:45:13 at > org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:106) > Feb 03 03:45:13 at > org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:94) > Feb 03 03:45:13 at > org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112) > Feb 03 03:45:13 at > java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604) > Feb 03 03:45:13 ... 3 more > {code} > https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=30662&view=logs&j=fb37c667-81b7-5c22-dd91-846535e99a97&t=39a035c3-c65e-573c-fb66-104c66c28912&l=5782 -- This message was sent by Atlassian Jira (v8.20.1#820001)