[feat-144][kafka] 修复没有 offsetEnd 参数时导致的空指针异常

Author: FlechazoW

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

@@ -51,16 +51,21 @@ public T deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
         if (endPartition.contains(record.partition())) {
             return null;
         }
-        Long endOffset = specificEndOffsets.get(topicPartition);
-        if (record.offset() >= endOffset) {
-            endPartition.add(record.partition());
-            return null;
+        if (specificEndOffsets != null) {
+            Long endOffset = specificEndOffsets.get(topicPartition);
+            if (record.offset() >= endOffset) {
+                endPartition.add(record.partition());
+                return null;
+            }
         }
+
         return super.deserialize(record);
     }
 
     public boolean isEndOfStream(T nextElement) {
-        return super.isEndOfStream(nextElement)
-                || endPartition.size() == specificEndOffsets.size();
+        boolean isEnd =
+                specificEndOffsets != null
+                        && endPartition.size() == specificEndOffsets.size();
+        return super.isEndOfStream(nextElement) || isEnd;
     }
 }