[FLINK-18241] Custom OptionsFactory in user code not working when configured via flink-conf.yaml - ASF JIRA

Details

Type: Bug
Status: Closed
Priority: Major
Resolution: Fixed
Affects Version/s: 1.10.0, 1.10.1
Fix Version/s: 1.10.2, 1.11.0
Component/s: Runtime / Configuration, Runtime / State Backends
Labels:
- pull-request-available

Description

It seems like Flink 1.10 broke custom OptionsFactory definitions via the state.backend.rocksdb.options-factory configuration if the implementation resides in the user-code jar file. This is particularly bad to debug RocksDB issues since we disabled its (ever-growing) LOG file in ~~FLINK-15068~~.

If you look at the stack trace from the error below, you will notice, that StreamExecutionEnvironment is not provided with a user-code classloader and will us the one of its own class which is the parent loader that does not know about our OptionsFactory. This exact same code was working with Flink 1.9.3.

(I believe putting the custom OptionsFactory into a separate jar file inside Flink's lib folder may be a workaround but that should ideally not be needed).

2020-06-09 16:18:59,409 ERROR org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Could not start cluster entrypoint StandaloneJobClusterEntryPoint.
org.apache.flink.runtime.entrypoint.ClusterEntrypointException: Failed to initialize the cluster entrypoint StandaloneJobClusterEntryPoint.
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:192) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runClusterEntrypoint(ClusterEntrypoint.java:525) [flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.container.entrypoint.StandaloneJobClusterEntryPoint.main(StandaloneJobClusterEntryPoint.java:116) [flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
Caused by: org.apache.flink.util.FlinkException: Could not create the DispatcherResourceManagerComponent.
        at org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory.create(DefaultDispatcherResourceManagerComponentFactory.java:261) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:220) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$0(ClusterEntrypoint.java:174) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:173) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        ... 2 more
Caused by: org.apache.flink.util.FlinkRuntimeException: Could not retrieve the JobGraph.
        at org.apache.flink.runtime.dispatcher.runner.JobDispatcherLeaderProcessFactoryFactory.createFactory(JobDispatcherLeaderProcessFactoryFactory.java:57) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunnerFactory.createDispatcherRunner(DefaultDispatcherRunnerFactory.java:51) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory.create(DefaultDispatcherResourceManagerComponentFactory.java:196) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:220) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$0(ClusterEntrypoint.java:174) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:173) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        ... 2 more
Caused by: org.apache.flink.util.FlinkException: Could not create the JobGraph from the provided user code jar.
        at org.apache.flink.container.entrypoint.ClassPathJobGraphRetriever.retrieveJobGraph(ClassPathJobGraphRetriever.java:114) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.JobDispatcherLeaderProcessFactoryFactory.createFactory(JobDispatcherLeaderProcessFactoryFactory.java:55) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunnerFactory.createDispatcherRunner(DefaultDispatcherRunnerFactory.java:51) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory.create(DefaultDispatcherResourceManagerComponentFactory.java:196) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:220) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$0(ClusterEntrypoint.java:174) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:173) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        ... 2 more
Caused by: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: org.apache.flink.util.DynamicCodeLoadingException: Cannot find configured options factory class: com.ververica.DefaultConfigurableOptionsFactoryWithLog
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:335) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.OptimizerPlanEnvironment.getPipeline(OptimizerPlanEnvironment.java:80) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:108) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:58) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.container.entrypoint.ClassPathJobGraphRetriever.retrieveJobGraph(ClassPathJobGraphRetriever.java:104) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.JobDispatcherLeaderProcessFactoryFactory.createFactory(JobDispatcherLeaderProcessFactoryFactory.java:55) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunnerFactory.createDispatcherRunner(DefaultDispatcherRunnerFactory.java:51) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory.create(DefaultDispatcherResourceManagerComponentFactory.java:196) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:220) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$0(ClusterEntrypoint.java:174) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:173) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        ... 2 more
Caused by: org.apache.flink.util.FlinkRuntimeException: org.apache.flink.util.DynamicCodeLoadingException: Cannot find configured options factory class: com.ververica.DefaultConfigurableOptionsFactoryWithLog
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.<init>(RocksDBStateBackend.java:376) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.configure(RocksDBStateBackend.java:394) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:47) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:32) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.state.StateBackendLoader.loadStateBackendFromConfig(StateBackendLoader.java:154) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.loadStateBackend(StreamExecutionEnvironment.java:767) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.configure(StreamExecutionEnvironment.java:750) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.<init>(StreamExecutionEnvironment.java:218) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.<init>(StreamExecutionEnvironment.java:190) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamPlanEnvironment.<init>(StreamPlanEnvironment.java:38) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.createStreamExecutionEnvironment(StreamExecutionEnvironment.java:1871) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at java.util.Optional.orElseGet(Optional.java:267) ~[?:1.8.0_252]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment(StreamExecutionEnvironment.java:1859) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at com.ververica.flinktraining.exercises.troubleshoot.TroubledStreamingJobUtils.createConfiguredEnvironment(TroubledStreamingJobUtils.java:40) ~[?:?]
        at com.ververica.flinktraining.solutions.troubleshoot.TroubledStreamingJobSolution43.main(TroubledStreamingJobSolution43.java:40) ~[?:?]
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_252]
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_252]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_252]
        at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_252]
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:321) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.OptimizerPlanEnvironment.getPipeline(OptimizerPlanEnvironment.java:80) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:108) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:58) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.container.entrypoint.ClassPathJobGraphRetriever.retrieveJobGraph(ClassPathJobGraphRetriever.java:104) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.JobDispatcherLeaderProcessFactoryFactory.createFactory(JobDispatcherLeaderProcessFactoryFactory.java:55) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunnerFactory.createDispatcherRunner(DefaultDispatcherRunnerFactory.java:51) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory.create(DefaultDispatcherResourceManagerComponentFactory.java:196) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:220) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$0(ClusterEntrypoint.java:174) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:173) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        ... 2 more
Caused by: org.apache.flink.util.DynamicCodeLoadingException: Cannot find configured options factory class: com.ververica.DefaultConfigurableOptionsFactoryWithLog
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.configureOptionsFactory(RocksDBStateBackend.java:605) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.<init>(RocksDBStateBackend.java:370) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.configure(RocksDBStateBackend.java:394) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:47) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:32) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.state.StateBackendLoader.loadStateBackendFromConfig(StateBackendLoader.java:154) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.loadStateBackend(StreamExecutionEnvironment.java:767) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.configure(StreamExecutionEnvironment.java:750) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.<init>(StreamExecutionEnvironment.java:218) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.<init>(StreamExecutionEnvironment.java:190) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamPlanEnvironment.<init>(StreamPlanEnvironment.java:38) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.createStreamExecutionEnvironment(StreamExecutionEnvironment.java:1871) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at java.util.Optional.orElseGet(Optional.java:267) ~[?:1.8.0_252]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment(StreamExecutionEnvironment.java:1859) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at com.ververica.flinktraining.exercises.troubleshoot.TroubledStreamingJobUtils.createConfiguredEnvironment(TroubledStreamingJobUtils.java:40) ~[?:?]
        at com.ververica.flinktraining.solutions.troubleshoot.TroubledStreamingJobSolution43.main(TroubledStreamingJobSolution43.java:40) ~[?:?]
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_252]
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_252]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_252]
        at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_252]
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:321) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.OptimizerPlanEnvironment.getPipeline(OptimizerPlanEnvironment.java:80) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:108) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:58) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.container.entrypoint.ClassPathJobGraphRetriever.retrieveJobGraph(ClassPathJobGraphRetriever.java:104) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.JobDispatcherLeaderProcessFactoryFactory.createFactory(JobDispatcherLeaderProcessFactoryFactory.java:55) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunnerFactory.createDispatcherRunner(DefaultDispatcherRunnerFactory.java:51) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory.create(DefaultDispatcherResourceManagerComponentFactory.java:196) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:220) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$0(ClusterEntrypoint.java:174) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:173) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        ... 2 more
Caused by: java.lang.ClassNotFoundException: com.ververica.DefaultConfigurableOptionsFactoryWithLog
        at java.net.URLClassLoader.findClass(URLClassLoader.java:382) ~[?:1.8.0_252]
        at java.lang.ClassLoader.loadClass(ClassLoader.java:418) ~[?:1.8.0_252]
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352) ~[?:1.8.0_252]
        at java.lang.ClassLoader.loadClass(ClassLoader.java:351) ~[?:1.8.0_252]
        at java.lang.Class.forName0(Native Method) ~[?:1.8.0_252]
        at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_252]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.configureOptionsFactory(RocksDBStateBackend.java:594) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.<init>(RocksDBStateBackend.java:370) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.configure(RocksDBStateBackend.java:394) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:47) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:32) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.state.StateBackendLoader.loadStateBackendFromConfig(StateBackendLoader.java:154) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.loadStateBackend(StreamExecutionEnvironment.java:767) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.configure(StreamExecutionEnvironment.java:750) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.<init>(StreamExecutionEnvironment.java:218) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.<init>(StreamExecutionEnvironment.java:190) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamPlanEnvironment.<init>(StreamPlanEnvironment.java:38) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.createStreamExecutionEnvironment(StreamExecutionEnvironment.java:1871) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at java.util.Optional.orElseGet(Optional.java:267) ~[?:1.8.0_252]
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment(StreamExecutionEnvironment.java:1859) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at com.ververica.flinktraining.exercises.troubleshoot.TroubledStreamingJobUtils.createConfiguredEnvironment(TroubledStreamingJobUtils.java:40) ~[?:?]
        at com.ververica.flinktraining.solutions.troubleshoot.TroubledStreamingJobSolution43.main(TroubledStreamingJobSolution43.java:40) ~[?:?]
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_252]
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_252]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_252]
        at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_252]
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:321) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.OptimizerPlanEnvironment.getPipeline(OptimizerPlanEnvironment.java:80) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:108) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:58) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.container.entrypoint.ClassPathJobGraphRetriever.retrieveJobGraph(ClassPathJobGraphRetriever.java:104) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.JobDispatcherLeaderProcessFactoryFactory.createFactory(JobDispatcherLeaderProcessFactoryFactory.java:55) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunnerFactory.createDispatcherRunner(DefaultDispatcherRunnerFactory.java:51) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory.create(DefaultDispatcherResourceManagerComponentFactory.java:196) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:220) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$0(ClusterEntrypoint.java:174) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]
        at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:173) ~[flink-dist_2.12-1.10.1-stream1.jar:1.10.1-stream1]

I attached the OptionsFactory that I used to show the error. (just put it into the user code jar and configure

state.backend.rocksdb.options-factory: com.ververica.DefaultConfigurableOptionsFactoryWithLog

Attachments

DefaultConfigurableOptionsFactoryWithLog.java
10/Jun/20 11:07
0.9 kB
Nico Kruber

Issue Links

Add Link

links to

GitHub Pull Request #12601

Delete this link

GitHub Pull Request #12607

Delete this link

Activity

Comment

This comment will be Viewable by All Users Viewable by All Users

Cancel

Custom OptionsFactory in user code not working when configured via flink-conf.yaml

Details

Description

Attachments

Attachments

Issue Links

Activity

People

Dates

Agile

Slack

Issue deployment