[fix][connector-rocketmq]Fix a NPE problem when checkpoint.interval is set too small(#6624) (#6625)

Author: YalikWang

Diff for: seatunnel-connectors-v2/connector-rocketmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceReader.java

@@ -242,17 +242,21 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
                 Long offset = entry.getValue();
                 try {
                     if (messageQueue != null && offset != null) {
-                        consumerThreads
-                                .get(messageQueue)
-                                .getTasks()
-                                .put(
-                                        consumer -> {
-                                            if (this.metadata.isEnabledCommitCheckpoint()) {
-                                                consumer.getOffsetStore()
-                                                        .updateOffset(messageQueue, offset, false);
-                                                consumer.getOffsetStore().persist(messageQueue);
-                                            }
-                                        });
+                        RocketMqConsumerThread rocketMqConsumerThread =
+                                consumerThreads.get(messageQueue);
+                        if (rocketMqConsumerThread != null) {
+                            rocketMqConsumerThread
+                                    .getTasks()
+                                    .put(
+                                            consumer -> {
+                                                if (this.metadata.isEnabledCommitCheckpoint()) {
+                                                    consumer.getOffsetStore()
+                                                            .updateOffset(
+                                                                    messageQueue, offset, false);
+                                                    consumer.getOffsetStore().persist(messageQueue);
+                                                }
+                                            });
+                        }
                     }
                 } catch (InterruptedException e) {
                     log.error("commit offset failed", e);