danny0405 commented on code in PR #12166:
URL: https://github.com/apache/hudi/pull/12166#discussion_r1826399863
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadMonitoringFunction.java:
##########
@@ -143,6 +150,48 @@ public void testConsumeFromLastCommit() throws Exception {
}
}
+ @Test
+ public void testConsumeForSpeedLimitWhenEmptyCommitExists() throws Exception
{
+ // Step1 : create 4 empty commit
+ Configuration conf = new Configuration(this.conf);
+ conf.set(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_COPY_ON_WRITE);
+ conf.setBoolean(HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key(), true);
+
+ TestData.writeData(Collections.EMPTY_LIST, conf);
+ TestData.writeData(Collections.EMPTY_LIST, conf);
+ TestData.writeData(Collections.EMPTY_LIST, conf);
+ TestData.writeData(Collections.EMPTY_LIST, conf);
+
+ HoodieTableMetaClient metaClient =
HoodieTestUtils.init(conf.get(FlinkOptions.PATH),
HoodieTableType.COPY_ON_WRITE);
+ HoodieTimeline commitsTimeline = metaClient.reloadActiveTimeline()
+ .filter(hoodieInstant ->
hoodieInstant.getAction().equals(HoodieTimeline.COMMIT_ACTION));
+ HoodieInstant firstInstant = commitsTimeline.firstInstant().get();
+
+ // Step2: trigger streaming read from first instant and set
READ_COMMITS_LIMIT 2
+ conf.set(FlinkOptions.READ_AS_STREAMING, true);
+ conf.set(FlinkOptions.READ_STREAMING_SKIP_CLUSTERING, true);
+ conf.set(FlinkOptions.READ_STREAMING_SKIP_COMPACT, true);
+ conf.set(FlinkOptions.READ_COMMITS_LIMIT, 2);
+ conf.set(FlinkOptions.READ_START_COMMIT,
String.valueOf((Long.valueOf(firstInstant.getTimestamp()) - 100)));
+ StreamReadMonitoringFunction function = TestUtils.getMonitorFunc(conf);
+ try (AbstractStreamOperatorTestHarness<MergeOnReadInputSplit> harness =
createHarness(function)) {
+ harness.setup();
+ harness.open();
+
+ CountDownLatch latch = new CountDownLatch(0);
+ CollectingSourceContext sourceContext = new
CollectingSourceContext(latch);
+ function.monitorDirAndForwardSplits(sourceContext);
+ assertEquals(0, sourceContext.splits.size(), "There should be no
inputSplits");
+
+ // Step3: assert current IssuedOffset couldn't be null.
+ // Base on "IncrementalInputSplits#inputSplits =>
.startCompletionTime(issuedOffset != null ? issuedOffset :
this.conf.getString(FlinkOptions.READ_START_COMMIT))"
+ // If IssuedOffset still was null, hudi would take
FlinkOptions.READ_START_COMMIT again, which means streaming read is blocked.
+ assertTrue(function.getIssuedOffset() != null);
Review Comment:
Use assertNotNull
--
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]
