feat(client): enrich write commit callback message and fire it for table-service commits #18988
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,12 +20,17 @@ | |
|
|
||
| import org.apache.hudi.avro.model.HoodieCleanMetadata; | ||
| import org.apache.hudi.callback.HoodieClientInitCallback; | ||
| import org.apache.hudi.callback.HoodieWriteCommitCallback; | ||
| import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage; | ||
| import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage.PrevFilePaths; | ||
| import org.apache.hudi.callback.util.HoodieCommitCallbackFactory; | ||
| import org.apache.hudi.client.embedded.EmbeddedTimelineServerHelper; | ||
| import org.apache.hudi.client.embedded.EmbeddedTimelineService; | ||
| import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient; | ||
| import org.apache.hudi.client.transaction.TransactionManager; | ||
| import org.apache.hudi.client.utils.TransactionUtils; | ||
| import org.apache.hudi.common.engine.HoodieEngineContext; | ||
| import org.apache.hudi.common.model.HoodieBaseFile; | ||
| import org.apache.hudi.common.model.HoodieCommitMetadata; | ||
| import org.apache.hudi.common.model.HoodieWriteStat; | ||
| import org.apache.hudi.common.table.HoodieTableMetaClient; | ||
|
|
@@ -34,6 +39,7 @@ | |
| import org.apache.hudi.common.table.timeline.TimeGenerator; | ||
| import org.apache.hudi.common.table.timeline.TimeGenerators; | ||
| import org.apache.hudi.common.table.timeline.TimelineUtils; | ||
| import org.apache.hudi.common.table.view.TableFileSystemView.BaseFileOnlyView; | ||
| import org.apache.hudi.common.util.Option; | ||
| import org.apache.hudi.common.util.ReflectionUtils; | ||
| import org.apache.hudi.common.util.StringUtils; | ||
|
|
@@ -87,6 +93,14 @@ public abstract class BaseHoodieClient implements Serializable, AutoCloseable { | |
| protected final TransactionManager txnManager; | ||
| protected final TimeGenerator timeGenerator; | ||
|
|
||
| /** | ||
| * Lazily-initialized commit callback (HoodieWriteCommitCallback). Lifted from | ||
| * {@link BaseHoodieWriteClient} so that {@link BaseHoodieTableServiceClient} can also | ||
| * fire callbacks for compaction and clustering completions. Transient is fine | ||
| * because the callback is only ever invoked from the driver after a commit. | ||
| */ | ||
| protected transient HoodieWriteCommitCallback commitCallback; | ||
|
|
||
| /** | ||
| * Timeline Server has the same lifetime as that of Client. Any operations done on the same timeline service will be | ||
| * able to take advantage of the cached file-system view. New completed actions will be synced automatically in an | ||
|
|
@@ -462,4 +476,78 @@ 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 fireCommitCallback(String commitTime, | ||
|
Contributor
There was a problem hiding this comment. mayFireCommitCallback or fireCommitCallbackIfNecessary |
||
| String commitActionType, | ||
| List<HoodieWriteStat> stats, | ||
| BaseFileOnlyView fsView, | ||
| 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, fsView), | ||
|
Contributor
There was a problem hiding this comment. not sure the usage of the prev file paths, is it for debugging?
Member
Author
There was a problem hiding this comment. this was needed for an internal feature.. it lets callback consumers (e.g. testing frameworks, lineage/audit sinks) see which base file each update replaced without rebuilding a FileSystemView themselves. |
||
| 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 #fireCommitCallback} 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) { | ||
|
Contributor
There was a problem hiding this comment. query the fsview is costly, can we make it a supplier so that it is only called when necessary.
Member
Author
There was a problem hiding this comment. switched to a Supplier so the view is only resolved after the callback-enabled check, keeping the default (callback-off) path free. Even before, on the enabled path, it reuses the same cached view the writer already populated, so no extra I/O.
Contributor
There was a problem hiding this comment. what I'm saying is make the whole |
||
| 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() | ||
|
Contributor
There was a problem hiding this comment. 🤖 nit: - AI-generated; verify before applying. React 👍/👎 to flag quality. |
||
| ? prev.get().getBootstrapBaseFile().get().getPath() | ||
| : null; | ||
| out.put(stat.getFileId(), new PrevFilePaths(prevPath, bootstrapPath)); | ||
| } | ||
| return out; | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🤖 nit: the phrase "Mirrors the optional shape of
extraMetadata" is misleading —extraContextisMap<String, String>, notOption<Map<String, String>>, so the shapes are different. Could you drop or rephrase that sentence?- AI-generated; verify before applying. React 👍/👎 to flag quality.