wombatu-kun commented on code in PR #19202:
URL: https://github.com/apache/hudi/pull/19202#discussion_r3586536708
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestHoodieSourceSplitReader.java:
##########
@@ -512,25 +593,54 @@ public TestSplitReaderFunction(List<String> testData) {
}
@Override
- public RecordsWithSplitIds<HoodieRecordWithPosition<String>>
read(HoodieSourceSplit split) {
- readCount++;
+ public void open(HoodieSourceSplit split) {
+ openCount++;
lastReadSplit = split;
- ClosableIterator<String> iterator = createClosableIterator(testData);
- return BatchRecords.forRecords(
- split.splitId(),
- iterator,
- split.getFileOffset(),
- split.getConsumed()
- );
+ cursor = testData.iterator();
+ long consumed = split.getConsumed();
+ for (long i = 0; i < consumed; i++) {
+ if (cursor.hasNext()) {
+ cursor.next();
+ } else {
+ throw new IllegalStateException(
+ "Invalid starting record offset " + consumed + " for split " +
split.splitId());
+ }
+ }
+ nextRecordOffset = consumed;
+ }
+
+ @Override
+ public BatchRecords<String> readBatch(HoodieSourceSplit split, int
batchSize) {
+ List<String> buffer = new ArrayList<>();
+ while (buffer.size() < batchSize && cursor.hasNext()) {
+ buffer.add(cursor.next());
+ }
+ if (buffer.isEmpty()) {
+ return null;
+ }
+ long startingRecordOffset = nextRecordOffset;
+ nextRecordOffset += buffer.size();
+ return BatchRecords.forRecords(split.splitId(), buffer,
split.getFileOffset(), startingRecordOffset);
}
@Override
- public void close() throws Exception {
+ public void closeCurrentSplit() {
+ closeCurrentSplitCount++;
+ cursor = null;
+ }
+
+ @Override
+ public void close() {
closed = true;
}
+ // Number of splits opened; mirrors the old per-split read() count.
public int getReadCount() {
- return readCount;
+ return openCount;
+ }
Review Comment:
Done 77394e2
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
