[Bugifx][kafka] Fix kafka enumerator assign split NPE

[hailin0]

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.

[hailin0]

[Copilot]

Pull Request Overview

This PR fixes a potential NullPointerException in the Kafka enumerator's split assignment logic by adding initialization state checks.

• Introduces a volatile boolean flag (initialized) to control when discovery and assignment of splits should occur.
• Updates the discovery logic in open() and the registration logic in registerReader() to respect the new initialization state.

Comments suppressed due to low confidence (1)

seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceSplitEnumerator.java:262

• [nitpick] Adding 'initialized' to the condition in registerReader() ensures no premature split assignment. Please confirm that this check does not inadvertently delay processing of pending splits when readers are registered.

if (!pendingSplit.isEmpty() && initialized) {

[zhilinli123]

+1