nsivabalan commented on code in PR #12369:
URL: https://github.com/apache/hudi/pull/12369#discussion_r1863749522
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/AbstractIndexingCatchupTask.java:
##########
@@ -147,22 +155,37 @@ public void run() {
* If not, it waits until the instant is completed.
*
* @param instant HoodieInstant to check
- * @return null if instant is already caught up, else the instant after it
is completed.
+ * @return True if instant is already caught up, or no heartbeat, or expired
heartbeat. If heartbeat exists and not expired, then return false.
*/
- HoodieInstant awaitInstantCaughtUp(HoodieInstant instant) {
+ boolean awaitInstantCaughtUp(HoodieInstant instant) {
if (!metadataCompletedInstants.isEmpty() &&
metadataCompletedInstants.contains(instant.requestedTime())) {
currentCaughtupInstant = instant.requestedTime();
- return null;
+ return true;
}
if (!instant.isCompleted()) {
+ // check heartbeat
+ try {
+ // if no heartbeat, then ignore this instant
+ if (!HoodieHeartbeatClient.heartbeatExists(metaClient.getStorage(),
metaClient.getBasePath().toString(), instant.requestedTime())) {
Review Comment:
can we add java docs here.
essentially the notes that we discussed.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexingCatchupTask.java:
##########
@@ -135,6 +171,41 @@ public void testTaskInterrupted() {
}
}
+ /**
+ * Test case to cover heartbeat expiry. Validate that awaitInstantCaughtUp
+ * returns null when heartbeat has expired for the given instant.
+ */
+ @Test
+ public void testHeartbeatExpired() throws IOException {
+ HoodieInstant expiredInstant =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED, "commit",
"002");
+ HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
+ .withPath("/some/path")
+
.withCleanConfig(HoodieCleanConfig.newBuilder().withFailedWritesCleaningPolicy(HoodieFailedWritesCleaningPolicy.LAZY).build())
+ .build();
+ // Simulate heartbeat exists and expired
+ when(table.getConfig()).thenReturn(writeConfig);
+ HoodieStorage storage = mock(HoodieStorage.class);
+ when(metaClient.getStorage()).thenReturn(storage);
+ when(metaClient.getBasePath()).thenReturn(new StoragePath("/some/path"));
+ when(storage.exists(any())).thenReturn(true);
+ when(heartbeatClient.isHeartbeatExpired("002")).thenReturn(true);
Review Comment:
do we have a case where heart beat does not exist?
--
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]
