codope commented on code in PR #18988:
URL: https://github.com/apache/hudi/pull/18988#discussion_r3663285087
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java:
##########
@@ -462,4 +477,79 @@ private static Map<String, String>
collectRollingMetadataFromTimeline(
protected Option<Map<String, String>> updateExtraMetadata(Option<Map<String,
String>> extraMetadata) {
return CommitMetadataProperties.enrich(extraMetadata, config, context);
}
+
+ /**
+ * Fire {@link HoodieWriteCommitCallback} for a commit, if enabled. Shared by
+ * {@link BaseHoodieWriteClient#postCommit} (regular auto- and
explicit-commit paths)
+ * and {@link BaseHoodieTableServiceClient} (compaction and clustering
completions).
+ * Lazily constructs the callback instance from {@code
hoodie.write.commit.callback.class}.
+ *
+ * <p>Best-effort: catches and logs any exception from the user-supplied
callback so a
+ * misbehaving observer cannot fail the commit.
+ */
+ protected void fireCommitCallbackIfNecessary(String commitTime,
+ String commitActionType,
+ List<HoodieWriteStat> stats,
+ Supplier<BaseFileOnlyView>
fsViewSupplier,
+ Option<Map<String, String>>
extraMetadata) {
+ if (!config.writeCommitCallbackOn()) {
+ return;
+ }
+ try {
+ if (commitCallback == null) {
+ commitCallback = HoodieCommitCallbackFactory.create(config);
+ }
+ commitCallback.call(new HoodieWriteCommitCallbackMessage(
+ commitTime, config.getTableName(), config.getBasePath(),
+ stats, Option.of(commitActionType), extraMetadata,
+ resolvePrevFilePaths(stats, fsViewSupplier.get()),
+ Collections.emptyMap()));
+ } catch (Exception e) {
+ log.warn("HoodieWriteCommitCallback failed for commit {} ({}); ignoring",
+ commitTime, commitActionType, e);
+ }
+ }
+
+ /**
+ * Pre-resolve the previous base file (and bootstrap base file, if any) for
every
+ * {@link HoodieWriteStat} that represents an update, using a populated
+ * {@link BaseFileOnlyView}. The lookup is O(1) per stat against the cached
view, so
+ * this adds no I/O on top of what the writer already paid.
+ *
+ * <p>Used by {@link #fireCommitCallbackIfNecessary} call sites so the
callback message ships
+ * actual file paths rather than forcing each callback impl to rebuild a
+ * {@code FileSystemView}.
+ */
+ protected static Map<String, PrevFilePaths>
resolvePrevFilePaths(List<HoodieWriteStat> stats,
+
BaseFileOnlyView fsView) {
+ Map<String, PrevFilePaths> out = new HashMap<>();
+ if (stats == null || fsView == null) {
+ return out;
+ }
+ for (HoodieWriteStat stat : stats) {
+ String prevCommit = stat.getPrevCommit();
+ if (StringUtils.isNullOrEmpty(prevCommit) ||
HoodieWriteStat.NULL_COMMIT.equals(prevCommit)) {
+ continue;
+ }
+ Option<HoodieBaseFile> prev;
+ try {
+ prev = fsView.getBaseFileOn(stat.getPartitionPath(), prevCommit,
stat.getFileId());
+ } catch (Exception e) {
+ // Best-effort: a remote view 4xx/5xx, a stale view, or a replaced
file group must not
+ // fail the commit. Drop the prev path for this stat and keep going.
+ log.warn("Could not resolve prev base file for fileId={}
prevCommit={}; skipping",
+ stat.getFileId(), prevCommit, e);
+ continue;
+ }
+ if (!prev.isPresent()) {
+ continue;
+ }
+ String prevPath = prev.get().getPath();
+ String bootstrapPath = prev.get().getBootstrapBaseFile().isPresent()
Review Comment:
cached `prev.get()` and `getBootstrapBaseFile()` in locals
--
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]
