[FLINK-33577][dist] Change the default config file to config.yaml in flink-dist.

[JunRuiLee]

What is the purpose of the change

Change the default config file to config.yaml in flink-dist.

Brief change log

• Change the default config file to config.yaml in flink-dist.
• Update the content 'flink-conf.yaml' in the code to use 'config.yaml'

Verifying this change

This change is already covered by existing tests, include all e2e cases and it cases.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (yes / no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
• The serializers: (yes / no / don't know)
• The runtime per-record code paths (performance sensitive): (yes / no / don't know)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
• The S3 file system connector: (yes / no / don't know)

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

[flinkbot]

CI report:

• 77182d8 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[JunRuiLee]

Hi @HuangXingBo , this pr includes some PyFlink changes related to the adoption of a new configuration file, config.yaml. Could you please help review it?

[JunRuiLee]

Thanks @zhuzhurk for reviews, I've updated this pr accordingly, PTAL.

[HuangXingBo]

@JunRuiLee Thanks a lot for the great work. I have left some comments

[HuangXingBo]

I feel the logic of this block of code can be refactored:

if  flink-conf.yaml exist:
    yaml.parse
elif conf.yaml exist:
    old parse logic
else
    return default value

[HuangXingBo]

isStandardYaml -> is_standard_yaml is more pythonic

[HuangXingBo]

I feel the logic of this block of code can be refactored.

if key in [jars_key, classpaths_key]:
	jar_urls = parse_jars_value(value)
	add_jars_to_context_class_loader(jar_urls)

def parse_jars_value(value: str):
	is_standard_yaml = jvm.org.apache.flink.configuration.GlobalConfiguration.isStandardYaml()
    if is_standard_yaml:
    	import yaml
        jar_urls_list = yaml.safe_load(value)
        if isinstance(jar_urls_list, list):
        	return jar_urls_list

    return value

[HuangXingBo]

isStandardYaml->is_standard_yaml

[HuangXingBo]

I feel the logic of this block of code can be refactored:

def _add_jars_to_j_env_config(self, config_key):
	jar_urls = self.get_config().get(config_key, None)

	if jar_urls:
		jvm = get_gateway().jvm
		jar_urls_list = []
		_parse_urls([jvm.java.net.URL(url).toString() if item else "" for item in parse_jars_value(jar_urls, jvm)], jar_urls_list)

                  j_configuration = get_j_env_configuration(self._get_j_env())
                  _parse_urls(parse_jars_value(j_configuration.getString(config_key, ""), jvm), jar_urls_list)

                 j_configuration.setString(config_key, ";".join(jar_urls_list))

def _parse_urls(self, jar_urls, jar_urls_list):
    for url in jar_urls.split(";"):
    	url = url.strip()
    	if url != "" and url not in jar_urls_list:
        	jar_urls_list.append(url)

[JunRuiLee]

Thanks @HuangXingBo for reviews, I've updated this pr follow your comments, PTAL.

[HuangXingBo]

@JunRuiLee LGTM

[zhuzhurk]

LGTM.
The CI gives green. I will squash and merge the commits locally.