[FLINK-38918][PyFlink] Support per cluster offset for DynamicKafkaSource in pyflink

[KenanAdel]

What is the purpose of the change

This PR adds support for per-cluster starting and stopping offset initializers in PyFlink's DynamicKafkaSource, aligning the Python API with the underlying Java implementation introduced in FLINK-38876.

Brief change log

• Updated ClusterMetadata and SingleClusterTopicMetadataService constructors in dynamic_kafka.py to accept optional starting_offsets_initializer and stopping_offsets_initializer.
• Forwarded the PyFlink offset initializers to the corresponding Java objects via the Py4J gateway using _j_initializer.
• Ensured proper propagation of offset configuration from Python layer to the underlying Java metadata service.
• Added comprehensive unit tests in test_dynamic_kafka.py to verify offset forwarding and ensure backward compatibility when offsets are not provided.

Verifying this change

This change is covered by the following newly added unit tests in test_dynamic_kafka.py:

• test_single_cluster_metadata_service_with_offsets
• test_single_cluster_metadata_service_default_offsets
• test_cluster_metadata_with_offsets

[flinkbot]

CI report:

• Unknown: CANCELED
• 196032e Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

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

[KenanAdel]

@flinkbot run azure