[Bugfix][Kafka] In kafka flow mode, stop offse should be Long.MAX_VALUE

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

@@ -104,15 +104,22 @@ public SourceReader<SeaTunnelRow, KafkaSourceSplit> createReader(
     @Override
     public SourceSplitEnumerator<KafkaSourceSplit, KafkaSourceState> createEnumerator(
             SourceSplitEnumerator.Context<KafkaSourceSplit> enumeratorContext) {
-        return new KafkaSourceSplitEnumerator(kafkaSourceConfig, enumeratorContext, null);
+        return new KafkaSourceSplitEnumerator(
+                kafkaSourceConfig,
+                enumeratorContext,
+                null,
+                getBoundedness() == Boundedness.UNBOUNDED);
     }
 
     @Override
     public SourceSplitEnumerator<KafkaSourceSplit, KafkaSourceState> restoreEnumerator(
             SourceSplitEnumerator.Context<KafkaSourceSplit> enumeratorContext,
             KafkaSourceState checkpointState) {
         return new KafkaSourceSplitEnumerator(
-                kafkaSourceConfig, enumeratorContext, checkpointState);
+                kafkaSourceConfig,
+                enumeratorContext,
+                checkpointState,
+                getBoundedness() == Boundedness.UNBOUNDED);
     }
 
     @Override

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

@@ -37,6 +37,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -70,17 +71,21 @@ public class KafkaSourceSplitEnumerator
 
     private final Map<String, TablePath> topicMappingTablePathMap = new HashMap<>();
 
+    private boolean isStreamingMode;
+
     KafkaSourceSplitEnumerator(
             KafkaSourceConfig kafkaSourceConfig,
             Context<KafkaSourceSplit> context,
-            KafkaSourceState sourceState) {
+            KafkaSourceState sourceState,
+            boolean isStreamingMode) {
         this.kafkaSourceConfig = kafkaSourceConfig;
         this.tablePathMetadataMap = kafkaSourceConfig.getMapMetadata();
         this.context = context;
         this.assignedSplit = new HashMap<>();
         this.pendingSplit = new HashMap<>();
         this.adminClient = initAdminClient(this.kafkaSourceConfig.getProperties());
         this.discoveryIntervalMillis = kafkaSourceConfig.getDiscoveryIntervalMillis();
+        this.isStreamingMode = isStreamingMode;
     }
 
     @VisibleForTesting
@@ -305,7 +310,9 @@ private Set<KafkaSourceSplit> getTopicInfo() throws ExecutionException, Interrup
                             // Obtain the corresponding topic TablePath from kafka topic
                             TablePath tablePath = topicMappingTablePathMap.get(partition.topic());
                             KafkaSourceSplit split = new KafkaSourceSplit(tablePath, partition);
-                            split.setEndOffset(latestOffsets.get(split.getTopicPartition()));
+                            split.setEndOffset(
+                                    latestOffsets.getOrDefault(
+                                            split.getTopicPartition(), Long.MAX_VALUE));
                             return split;
                         })
                 .collect(Collectors.toSet());
@@ -344,6 +351,11 @@ private static int getSplitOwner(TopicPartition tp, int numReaders) {
     private Map<TopicPartition, Long> listOffsets(
             Collection<TopicPartition> partitions, OffsetSpec offsetSpec)
             throws ExecutionException, InterruptedException {
+
+        if (isStreamingMode) {
+            return Collections.emptyMap();
+        }
+
         Map<TopicPartition, OffsetSpec> topicPartitionOffsets =
                 partitions.stream()
                         .collect(Collectors.toMap(partition -> partition, __ -> offsetSpec));