[FLINK-16149][hotfix][core] Use GlobalConfigurationLoader on StreamPlanEnvironment #31
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…() on StreamPlanEnvironment Because the statefun-flink-launcher is based on JobClusterEntrypoint it pulls the StreamGraph using a StreamPlanEnvironment. In this case, the StreamExecutionEnviornment configuration is set to an empty object. The only way to pull the actual flink-conf in this case is via GlobalConfiguration.loadConfiguration(). This is safe because StreamPlanEnvironment implies job based deployment. In all other cases the configuration is still pulled via reflection.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This is a follow up to #27
I noticed when running a container-based deployment that my stateful function-specific configurations were not being respected. Other flink configurations were properly set.
If you follow the execution path of the stateful-function-launcher, you will see that it calls
PackagedProgramUtils#createJobGraphwhich in turn sets anOptimizerPlanEnvironment. This is because it uses a Job-based deployment and the OptimizerPlanEnvironment pulls the job graph without executing the job so it can be embedded in the image.When this execution environment is set,
StreamExecutionEnvironment#getExecutionEnvironmentreturns aStreamPlanEnvironment. This subclass ofStreamExecutionEnvironmentcallsStreamExecutionEnvironment's no-op constructor which sets an empty configuration object.Because
StreamPlanEnvironmentimplies job-based deployment, we can safely callGlobalConfiguration#loadConfiguration. In all other cases we fall back to reflection.I have tested this using statefun images and deployed against a session cluster.
cc @igalshilman @tzulitai