[Fix][Connector-V2] Fix kafka database name

[corgy-w]

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config
• Update the release-note.

[Copilot]

Pull Request Overview

This pull request fixes an issue with the Kafka connector’s database naming by refactoring the table path construction logic. Key changes include:

• Extraction of the table path resolution into a new method (getTablePathFromSchema) for improved flexibility.
• Updates to the catalog table creation to use the new table path resolution.
• Addition of a new unit test (testDeserializationWithSchema) to validate deserialization behavior when a schema is provided.

Reviewed Changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfigTest.java	Added a unit test to verify deserialization with a provided schema and table identifier.
seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java	Refactored the table path construction by extracting logic into getTablePathFromSchema to fix database naming issues.

Comments suppressed due to low confidence (2)

seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java:265

• Ensure that the schema configuration key is used consistently across the connector. You might consider aligning the usage of KafkaSourceOptions.SCHEMA and TableSchemaOptions.SCHEMA to prevent potential misconfiguration.

readonlyConfig.getOptional(TableSchemaOptions.SCHEMA)

seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfigTest.java:86

• [nitpick] Consider clarifying the test table identifier format (e.g., separating database and table parts) to ensure it aligns with production expectations and avoids ambiguity.

schema.put(TABLE.key(), "test.test");

[hailin0]

LGTM

waiting for ci passed