lizhimins commented on code in PR #126:
URL: https://github.com/apache/rocketmq-flink/pull/126#discussion_r1982645357
##########
src/main/java/org/apache/flink/connector/rocketmq/source/reader/RocketMQSplitReader.java:
##########
@@ -103,13 +115,26 @@ public RocketMQSplitReader(
@Override
public RecordsWithSplitIds<MessageView> fetch() throws IOException {
+ lock.lock();
wakeup = false;
RocketMQRecordsWithSplitIds<MessageView> recordsWithSplitIds =
new RocketMQRecordsWithSplitIds<>(rocketmqSourceReaderMetrics);
+ try {
+ this.recordsWithSplitIds.finishedSplits.forEach(
+ splitId -> recordsWithSplitIds.addFinishedSplit(splitId));
+ this.recordsWithSplitIds.finishedSplits.clear();
+ } finally {
+ lock.unlock();
+ }
try {
Duration duration =
Duration.ofMillis(this.configuration.getLong(RocketMQOptions.POLL_TIMEOUT));
List<MessageView> messageExtList = consumer.poll(duration);
+ LOG.info(
Review Comment:
这里打日志会不会日志太多
##########
src/main/java/org/apache/flink/connector/rocketmq/source/split/RocketMQPartitionSplitSerializer.java:
##########
@@ -61,8 +62,13 @@ public RocketMQSourceSplit deserialize(int version, byte[]
serialized) throws IO
int partition = in.readInt();
long startingOffset = in.readLong();
long stoppingOffset = in.readLong();
+ if (version == SNAPSHOT_VERSION) {
Review Comment:
这个判断是干什么用的
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
