[jira] [Created] (FLINK-30278) Unexpected config mutation in SinkTransformationTranslator

["Anton Kalashnikov (Jira)" <ji...@apache.org>]

Anton Kalashnikov created FLINK-30278:
-----------------------------------------

             Summary: Unexpected config mutation in SinkTransformationTranslator 
                 Key: FLINK-30278
                 URL: https://issues.apache.org/jira/browse/FLINK-30278
             Project: Flink
          Issue Type: Bug
          Components: Runtime / Configuration
    Affects Versions: 1.16.0, 1.17.0
            Reporter: Anton Kalashnikov


If we forbid changing configuration programmatically(`execution.program-config.enabled`) and try to use `FileSink`. The following exception will occur:
<noformat>
  org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Not allowed configuration change(s) were detected:
 - Configuration parallelism.default:1 not allowed in the configuration object ExecutionConfig.
   at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:364)
   at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:225)
   at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:98)
   at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.runApplicationEntryPoint(ApplicationDispatcherBootstrap.java:319)
   at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.lambda$runApplicationAsync$2(ApplicationDispatcherBootstrap.java:262)
   at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
   at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
   at org.apache.flink.runtime.concurrent.akka.ActorSystemScheduledExecutorAdapter$ScheduledFutureTask.run(ActorSystemScheduledExecutorAdapter.java:171)
   at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)
   at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$withContextClassLoader$0(ClassLoadingUtils.java:41)
   at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:49)
   at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:48)
   at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source)
   at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown Source)
   at java.base/java.util.concurrent.ForkJoinPool.scan(Unknown Source)
   at java.base/java.util.concurrent.ForkJoinPool.runWorker(Unknown Source)
   at java.base/java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source)
Caused by: org.apache.flink.client.program.MutatedConfigurationException: Not allowed configuration change(s) were detected:
 - Configuration parallelism.default:1 not allowed in the configuration object ExecutionConfig.
   at org.apache.flink.client.program.StreamContextEnvironment.checkNotAllowedConfigurations(StreamContextEnvironment.java:235)
   at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:175)
   at org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:115)
   at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:2049)
   at org.apache.flink.streaming.examples.wordcount.WordCount.main(WordCount.java:81)
   at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
   at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
   at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
   at java.base/java.lang.reflect.Method.invoke(Unknown Source)
   at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:347)
   ... 16 more
<noformat>

It happens since inside of `SinkTransformationTranslator` we have following logic:
* Remeber the current parallelism
* Set parallelism to default
* Do transformation
* Set parallelism to remembered one

But if the initial prallelism is default we actually should do nothing but according current logic we explicitly set default value to the configuration which actually is the programmatic config mutation(which we want to avoid)

See org.apache.flink.streaming.runtime.translators.SinkTransformationTranslator.SinkExpander#executionEnvironment:341 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)