[hotfix-38914][kafka] Fix irrational logic of kafka seekOffset.

Signed-off-by: tiezhu <[email protected]>

Author: FlechazoW

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/deserialization/DtKafkaDeserializationSchemaWrapper.java

@@ -66,7 +66,7 @@ public T deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
         if (endPartition.contains(record.partition())) {
             return null;
         }
-        if (specificEndOffsets != null) {
+        if (specificEndOffsets != null && !specificEndOffsets.isEmpty()) {
             Long endOffset = specificEndOffsets.get(topicPartition);
             if (endOffset != null && record.offset() >= endOffset - 1) {
                 endPartition.add(record.partition());

kafka/kafka-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer.java

@@ -112,7 +112,7 @@ public void run(SourceFunction.SourceContext<Row> sourceContext) throws Exceptio
                                                     MetricGroup consumerMetricGroup,
                                                     boolean useMetrics) throws Exception {
 
-        final OffsetMap offsetMap = seekOffset(props, topic);
+        final OffsetMap offsetMap = sampleSize > 0 ? seekOffset(props, topic) : new OffsetMap();
         Map<KafkaTopicPartition, Long> rebuild;
 
         rebuild =

kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer09.java

@@ -113,7 +113,7 @@ public void run(SourceFunction.SourceContext<Row> sourceContext) throws Exceptio
                                                     MetricGroup consumerMetricGroup,
                                                     boolean useMetrics) throws Exception {
 
-        final OffsetMap offsetMap = seekOffset(props, topic);
+        final OffsetMap offsetMap = sampleSize > 0 ? seekOffset(props, topic) : new OffsetMap();
         Map<KafkaTopicPartition, Long> rebuild;
 
         rebuild =

kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer010.java

@@ -111,7 +111,7 @@ public void run(SourceContext<Row> sourceContext) throws Exception {
                                                     OffsetCommitMode offsetCommitMode,
                                                     MetricGroup consumerMetricGroup,
                                                     boolean useMetrics) throws Exception {
-        final OffsetMap offsetMap = seekOffset(props, topic);
+        final OffsetMap offsetMap = sampleSize > 0 ? seekOffset(props, topic) : new OffsetMap();
         Map<KafkaTopicPartition, Long> rebuild;
 
         rebuild =

kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer011.java

@@ -111,7 +111,7 @@ public void run(SourceContext<Row> sourceContext) throws Exception {
                                                     OffsetCommitMode offsetCommitMode,
                                                     MetricGroup consumerMetricGroup,
                                                     boolean useMetrics) throws Exception {
-        final OffsetMap offsetMap = seekOffset(props, topic);
+        final OffsetMap offsetMap = sampleSize > 0 ? seekOffset(props, topic) : new OffsetMap();
         Map<KafkaTopicPartition, Long> rebuild;
 
         rebuild =