lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-918] Fix kafkaOffsetGen can not read kafka data bug (#1652)

Browse Source
This commit is contained in:
liujinhui
2020-06-08 20:46:47 +08:00
committed by GitHub
parent 2901f5423a
commit 97ab97b726
1 changed files with 6 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
View File

@@ -21,6 +21,7 @@ package org.apache.hudi.utilities.sources.helpers;
import org.apache.hudi.DataSourceUtils;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -207,6 +208,11 @@ public class KafkaOffsetGen {
maxEventsToReadFromKafka = (maxEventsToReadFromKafka == Long.MAX_VALUE || maxEventsToReadFromKafka == Integer.MAX_VALUE)
? Config.maxEventsFromKafkaSource : maxEventsToReadFromKafka;
long numEvents = sourceLimit == Long.MAX_VALUE ? maxEventsToReadFromKafka : sourceLimit;
if (numEvents < toOffsets.size()) {
throw new HoodieException("sourceLimit should not be less than the number of kafka partitions");
}
return CheckpointUtils.computeOffsetRanges(fromOffsets, toOffsets, numEvents);
}