[hotfix][cdc-composer] Fix "URI is not hierarchical" when resolving factory JAR path

[wangmingzhou1986]

What is the purpose of the change

FactoryDiscoveryUtils#getJarPathByIdentifier resolves the code source location of a
connector Factory to a local path in order to decide whether the containing JAR has to be
shipped to the TaskManagers. It does so with Paths.get(url.toURI()), which requires a
hierarchical URI.

A code source location is not always hierarchical. When a pipeline runs in application mode
and Flink picks up the JARs under usrlib, the user code class loader is built with relative
URLs, e.g. file:usrlib/flink-cdc-pipeline-connector-mysql-3.6.0-1.20.jar. Such a URL has no
authority and its path does not start with a slash, so the URI is opaque and
Paths.get(URI) throws:

java.lang.IllegalArgumentException: URI is not hierarchical
	at java.base/sun.nio.fs.UnixUriUtils.fromUri(Unknown Source)
	at java.base/java.nio.file.Paths.get(Unknown Source)
	at org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils.getJarPathByIdentifier(FactoryDiscoveryUtils.java:104)
	at org.apache.flink.cdc.composer.flink.translator.DataSourceTranslator.createDataSource(DataSourceTranslator.java:94)
	at org.apache.flink.cdc.composer.flink.FlinkPipelineComposer.translate(FlinkPipelineComposer.java:168)
	at org.apache.flink.cdc.composer.flink.FlinkPipelineComposer.compose(FlinkPipelineComposer.java:106)

which is wrapped into RuntimeException: Failed to search JAR by factory identifier "mysql"
and fails the job before it is ever submitted.

How it was observed

Running a CDC YAML pipeline as a Flink Kubernetes Operator FlinkDeployment, with the CDC
connector JARs placed under /opt/flink/usrlib. Printing the code source of each class from
inside the JobManager shows the difference clearly:

class	CodeSource location	URI#isOpaque
application entry class (came in via jarURI: local:///opt/flink/usrlib/...)	file:/opt/flink/usrlib/entry.jar	false
MySqlDataSourceFactory (picked up by the usrlib scan)	file:usrlib/flink-cdc-pipeline-connector-mysql-3.6.0-1.20.jar	true
PaimonDataSinkFactory	file:usrlib/flink-cdc-pipeline-connector-paimon-3.6.0-1.20.jar	true

Only the absoluteness of the URL differs; the packaging is identical. The existing
local:///opt/flink/usrlib/ short-circuit does not match these URLs, so execution reaches
Paths.get and throws.

Brief change log

• Extract the URL-to-path resolution into a small resolveLocalPath(URL) helper.
• The helper handles opaque URIs by resolving the scheme-specific part against the working
  directory, and returns an absolute path in both cases.
• If the location cannot be mapped to a local file at all, log a warning and return
  Optional.empty() (i.e. do not attempt to upload the JAR) instead of throwing.
• The returned URL is now normalised to an absolute one, so callers that use it for uploading
  no longer receive a working-directory-relative URL.

Behaviour for locations that were already hierarchical is unchanged: the same path is
resolved, the same isDirectory check applies, and an equivalent absolute file: URL is
returned.

Verifying this change

This change is a small bug fix in path resolution and is covered by the existing
flink-cdc-composer tests. It can be reproduced without this patch by running any CDC
pipeline in application mode with the connector JARs under usrlib - the job fails during
composition with the stack trace above; with the patch the JAR path resolves and composition
proceeds.

Does this pull request potentially affect one of the following parts

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: yes - it makes JAR resolution work in
  application mode where it previously threw; no configuration change is required.
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature: no
• If yes, how is the feature documented: not applicable