danny0405 commented on code in PR #9850:
URL: https://github.com/apache/hudi/pull/9850#discussion_r1358086955
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnReadWithCompact.java:
##########
@@ -70,6 +86,80 @@ protected Map<String, String> getMiniBatchExpected() {
return expected;
}
+ @Test
+ public void testLocklessMultiWriterWithPartialUpdatePayload() throws
Exception {
+ conf.setString(HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key(),
WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.name());
+ conf.setString(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.BUCKET.name());
+ conf.setString(FlinkOptions.PAYLOAD_CLASS_NAME,
PartialUpdateAvroPayload.class.getName());
+ // disable schedule compaction in writers
+ conf.setBoolean(FlinkOptions.COMPACTION_SCHEDULE_ENABLED, false);
+ conf.setBoolean(FlinkOptions.PRE_COMBINE, true);
+
+ // start pipeline1 and insert record: [id1,par1,id1,Danny,null,1,par1],
suspend the tx commit
+ List<RowData> dataset1 = Collections.singletonList(
+ insertRow(
+ StringData.fromString("id1"), StringData.fromString("Danny"), null,
+ TimestampData.fromEpochMillis(1), StringData.fromString("par1")));
+ TestHarness pipeline1 = preparePipeline(conf)
+ .consume(dataset1)
+ .assertEmptyDataFiles();
+
+ // start pipeline2 and insert record: [id1,par1,id1,null,23,1,par1],
suspend the tx commit
+ Configuration conf2 = conf.clone();
+ conf2.setString(FlinkOptions.WRITE_CLIENT_ID, "2");
+ List<RowData> dataset2 = Collections.singletonList(
+ insertRow(
+ StringData.fromString("id1"), null, 23,
+ TimestampData.fromEpochMillis(2), StringData.fromString("par1")));
+ TestHarness pipeline2 = preparePipeline(conf2)
+ .consume(dataset2)
+ .assertEmptyDataFiles();
+
+ // step to commit the 1st txn
Review Comment:
Can we add another case: we have both completed and inflight instants, then
we schedule a compaction, and we validate the ro and rt view for correctness.
--
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]
