liaoxin01 commented on code in PR #30136:
URL: https://github.com/apache/doris/pull/30136#discussion_r1460156120
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaRoutineLoadJob.java:
##########
@@ -734,15 +734,26 @@ public boolean hasMoreDataToConsume(UUID taskId,
Map<Integer, Long> partitionIdT
// check again
for (Map.Entry<Integer, Long> entry : partitionIdToOffset.entrySet()) {
- if (cachedPartitionWithLatestOffsets.containsKey(entry.getKey())
- && entry.getValue() <
cachedPartitionWithLatestOffsets.get(entry.getKey())) {
- LOG.debug("has more data to consume. offsets to be consumed:
{}, latest offsets: {}, task {}, job {}",
- partitionIdToOffset, cachedPartitionWithLatestOffsets,
taskId, id);
- return true;
+ Integer partitionId = entry.getKey();
+ if (cachedPartitionWithLatestOffsets.containsKey(partitionId)) {
+ long partitionLatestOffset =
cachedPartitionWithLatestOffsets.get(partitionId);
+ long recordPartitionOffset = entry.getValue();
+ if (recordPartitionOffset < partitionLatestOffset) {
+ LOG.debug("has more data to consume. offsets to be
consumed: {},"
+ + " latest offsets: {}, task {}, job {}",
+ partitionIdToOffset,
cachedPartitionWithLatestOffsets, taskId, id);
+ return true;
+ } else if (recordPartitionOffset > partitionLatestOffset) {
+ String msg = "offset set in job: " + recordPartitionOffset
+ + " is greater than kafka latest offset: "
+ + partitionLatestOffset + " partition id: "
+ + partitionId;
+ throw new UserException(msg);
+ }
}
}
- LOG.debug("no more data to consume. offsets to be consumed: {}, latest
offsets: {}, task {}, job {}",
+ LOG.info("no more data to consume. offsets to be consumed: {}, latest
offsets: {}, task {}, job {}",
Review Comment:
Will this log be printed frequently when no more data to consume？
--
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...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
