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

Author: hailin0

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

@@ -68,6 +68,7 @@ public class KafkaSourceSplitEnumerator
     private final Map<TopicPartition, KafkaSourceSplit> assignedSplit;
     private ScheduledExecutorService executor;
     private ScheduledFuture<?> scheduledFuture;
+    private volatile boolean initialized;
 
     private final Map<String, TablePath> topicMappingTablePathMap = new HashMap<>();
 
@@ -128,7 +129,9 @@ public void open() {
                     executor.scheduleWithFixedDelay(
                             () -> {
                                 try {
-                                    discoverySplits();
+                                    if (initialized) {
+                                        discoverySplits();
+                                    }
                                 } catch (Exception e) {
                                     log.error("Dynamic discovery failure:", e);
                                 }
@@ -144,6 +147,9 @@ public void run() throws ExecutionException, InterruptedException {
         fetchPendingPartitionSplit();
         setPartitionStartOffset();
         assignSplit();
+        if (!initialized) {
+            initialized = true;
+        }
     }
 
     private void setPartitionStartOffset() throws ExecutionException, InterruptedException {
@@ -254,7 +260,7 @@ public void handleSplitRequest(int subtaskId) {
 
     @Override
     public void registerReader(int subtaskId) {
-        if (!pendingSplit.isEmpty()) {
+        if (!pendingSplit.isEmpty() && initialized) {
             assignSplit();
         }
     }