[Fix] [Connector] Rocketmq source startOffset greater than endOffset …

…error (apache#6287)
ruanwenjun · Jan 29, 2024 · cd44b58 · cd44b58
1 parent 064dc29
commit cd44b58
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/.../apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumerator.java b/.../apache/seatunnel/connectors/seatunnel/rocketmq/source/RocketMqSourceSplitEnumerator.java
@@ -149,7 +149,10 @@ public void addSplitsBack(List<RocketMqSourceSplit> splits, int subtaskId) {
                             ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
             splits.forEach(
                     split -> {
-                        split.setStartOffset(split.getEndOffset() + 1);
+                        split.setStartOffset(
+                                Math.min(
+                                        split.getEndOffset() + 1,
+                                        listOffsets.get(split.getMessageQueue())));
                         split.setEndOffset(listOffsets.get(split.getMessageQueue()));
                     });
             return splits.stream()