You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2022/05/11 16:25:58 UTC
[GitHub] [hudi] codope commented on a diff in pull request #5557: [HUDI-4084 Adding support to test async table services with integ test suite framework
codope commented on code in PR #5557:
URL: https://github.com/apache/hudi/pull/5557#discussion_r870516851
##########
hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/dag/nodes/BaseValidateDatasetNode.java:
##########
@@ -157,6 +171,50 @@ public void execute(ExecutionContext context, int curItrCount) throws Exception
}
}
+ private void awaitUntilDeltaStreamerCaughtUp(ExecutionContext context, String hudiTablePath, FileSystem fs, String inputPath) throws IOException, InterruptedException {
+ HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(new Configuration(fs.getConf())).setBasePath(hudiTablePath).build();
+ HoodieTimeline commitTimeline = meta.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ Option<String> latestCheckpoint = getLatestCheckpoint(commitTimeline);
+ FileStatus[] subDirs = fs.listStatus(new Path(inputPath));
Review Comment:
this is going to add to the validation time but i guess there is no other simpler way.
##########
hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/dag/nodes/BaseValidateDatasetNode.java:
##########
@@ -157,6 +171,50 @@ public void execute(ExecutionContext context, int curItrCount) throws Exception
}
}
+ private void awaitUntilDeltaStreamerCaughtUp(ExecutionContext context, String hudiTablePath, FileSystem fs, String inputPath) throws IOException, InterruptedException {
+ HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(new Configuration(fs.getConf())).setBasePath(hudiTablePath).build();
+ HoodieTimeline commitTimeline = meta.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ Option<String> latestCheckpoint = getLatestCheckpoint(commitTimeline);
+ FileStatus[] subDirs = fs.listStatus(new Path(inputPath));
+ List<FileStatus> subDirList = Arrays.asList(subDirs);
+ subDirList.sort(Comparator.comparingLong(entry -> Long.parseLong(entry.getPath().getName())));
+ String latestSubDir = subDirList.get(subDirList.size() -1).getPath().getName();
+ log.info("Latest sub directory in input path " + latestSubDir + ", latest checkpoint from deltastreamer " +
+ (latestCheckpoint.isPresent() ? latestCheckpoint.get() : "none"));
+ long maxWaitTime = 5 * 60 * 1000;
Review Comment:
Will this wait be good enough for long running jobs as well? Maybe consider making it configurable.
##########
hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/dag/nodes/BaseValidateDatasetNode.java:
##########
@@ -157,6 +171,50 @@ public void execute(ExecutionContext context, int curItrCount) throws Exception
}
}
+ private void awaitUntilDeltaStreamerCaughtUp(ExecutionContext context, String hudiTablePath, FileSystem fs, String inputPath) throws IOException, InterruptedException {
+ HoodieTableMetaClient meta = HoodieTableMetaClient.builder().setConf(new Configuration(fs.getConf())).setBasePath(hudiTablePath).build();
+ HoodieTimeline commitTimeline = meta.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ Option<String> latestCheckpoint = getLatestCheckpoint(commitTimeline);
+ FileStatus[] subDirs = fs.listStatus(new Path(inputPath));
Review Comment:
maybe we can use the files index if metadata table is enabled and fallback to listStatus if not.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org