cshuo commented on code in PR #19973:
URL: https://github.com/apache/hudi/pull/19973#discussion_r4025822386
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java:
##########
@@ -1033,6 +1037,29 @@ public void testCacheCleanOfRecordIndexBackend() throws
Exception {
.checkWrittenData(EXPECTED1);
}
+ @Test
+ public void testIndexWriteFunctionWithSmallBuffer() throws Exception {
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.GLOBAL_RECORD_LEVEL_INDEX.name());
+
conf.setString(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key(),
"true");
+ conf.setString(HoodieMetadataConfig.STREAMING_WRITE_ENABLED.key(), "true");
+ conf.set(FlinkOptions.INDEX_RLI_WRITE_BUFFER_SIZE, 1L);
+
+ StreamWriteFunctionWrapper<RowData> pipeline =
+ (StreamWriteFunctionWrapper<RowData>)
TestData.getWritePipeline(tempFile.getAbsolutePath(), conf);
+ pipeline.openFunction();
+ try {
+ // The record cannot fit even after flushing and resetting the 1 MB
index buffer.
+ RowData indexRow = GenericRowData.of(IndexRowUtils.RLI_TYPE,
+ StringData.fromString("k".repeat(2 * 1024 * 1024)),
+ StringData.fromString("par1"), StringData.fromString("file1"));
+ HoodieException exception = assertThrows(HoodieException.class,
+ () -> pipeline.getIndexWriteFunction().processElement(indexRow,
null, null));
+ assertEquals("Index write buffer is too small to hold a single record.",
exception.getMessage());
Review Comment:
Fixed in 88121043a73. Added `testIndexWriteFunctionRetriesAfterBufferFull`,
which feeds individually fitting records through the full write pipeline until
the 1 MB index buffer triggers a flush. It checks that all earlier records were
flushed before any checkpoint and that the triggering record is retained in the
buffer. After checkpoint completion, it reads the index and verifies every key,
partition, and file ID.
--
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]
