This is an automated email from the ASF dual-hosted git repository. duhengforever pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/rocketmq-flink.git
commit 624994d402106d7858960c1e1b5bd9c05d66f823 Author: SuXingLee <[email protected]> AuthorDate: Fri Nov 23 10:06:20 2018 +0800 Fix NPE when getting offset(#153) rockemq-flink: RocketMQSource will throw a NullPointerException --- src/main/java/org/apache/rocketmq/flink/RocketMQSource.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/main/java/org/apache/rocketmq/flink/RocketMQSource.java b/src/main/java/org/apache/rocketmq/flink/RocketMQSource.java index 2dc8fd5..8e8e57b 100644 --- a/src/main/java/org/apache/rocketmq/flink/RocketMQSource.java +++ b/src/main/java/org/apache/rocketmq/flink/RocketMQSource.java @@ -235,9 +235,8 @@ public class RocketMQSource<OUT> extends RichParallelSourceFunction<OUT> throw new IllegalArgumentException("Unknown value for CONSUMER_OFFSET_RESET_TO."); } } - offsetTable.put(mq, offset); } - + offsetTable.put(mq, offset); return offsetTable.get(mq); }
