[feat-144][kafka] 修复因为specificEndOffsets 获取不到partition值导致的空指针异常

Author: FlechazoW

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

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