Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -17,15 +17,27 @@

package org.apache.hudi.callback;

import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage.PrevFilePaths;
import org.apache.hudi.common.model.BaseFile;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.table.view.TableFileSystemView.BaseFileOnlyView;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.exception.HoodieCommitCallbackException;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
* Util helps to prepare callback message.
*/
@Slf4j
public class HoodieWriteCommitCallbackUtil {

private static final ObjectMapper MAPPER = new ObjectMapper();
Expand All @@ -41,4 +53,46 @@ public static String convertToJsonString(Object obj) {
}
}

/**
* 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>Feeds {@link org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage#getPrevFilePaths()}
* so the callback message can ship actual file paths rather than forcing each callback
* impl to rebuild a {@code FileSystemView}.
*/
public static Map<String, PrevFilePaths> resolvePrevFilePaths(List<HoodieWriteStat> stats,
BaseFileOnlyView fsView) {
Map<String, PrevFilePaths> pathsByFileId = new HashMap<>();
if (stats == null || fsView == null) {
return pathsByFileId;
}
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;
}
HoodieBaseFile prevBaseFile = prev.get();
Option<BaseFile> bootstrapBaseFile = prevBaseFile.getBootstrapBaseFile();
String prevPath = prevBaseFile.getPath();
String bootstrapPath = bootstrapBaseFile.isPresent() ? bootstrapBaseFile.get().getPath() : null;
pathsByFileId.put(stat.getFileId(), new PrevFilePaths(prevPath, bootstrapPath));
}
return pathsByFileId;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -19,20 +19,26 @@

import org.apache.hudi.ApiMaturityLevel;
import org.apache.hudi.PublicAPIClass;
import org.apache.hudi.callback.HoodieWriteCommitCallbackUtil;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.table.view.TableFileSystemView.BaseFileOnlyView;
import org.apache.hudi.common.util.Lazy;
import org.apache.hudi.common.util.Option;

import lombok.AllArgsConstructor;
import lombok.AccessLevel;
import lombok.Getter;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;

/**
* Base callback message, which contains commitTime and tableName only for now.
*/
@AllArgsConstructor
@Getter
@PublicAPIClass(maturity = ApiMaturityLevel.EVOLVING)
public class HoodieWriteCommitCallbackMessage implements Serializable {
Expand Down Expand Up @@ -69,10 +75,116 @@ public class HoodieWriteCommitCallbackMessage implements Serializable {
*/
private final Option<Map<String, String>> extraMetadata;

/**
* Previous base file paths keyed by fileId, derived from {@link #hoodieWriteStat} and the
* {@link BaseFileOnlyView} handed over by the write client, so that callback
* implementations don't have to rebuild a view themselves. Empty for inserts and for
* callers that don't supply a view.
*
* <p>Holds the resolved map once {@link #getPrevFilePaths()} has run, and stays null until
* then. Not transient: this is the copy that crosses Java serialization, which is why
* {@link #writeObject} forces resolution before writing. Excluded from the generated
* getters so it is published only through {@link #getPrevFilePaths()}.
*/
@Getter(AccessLevel.NONE)
private volatile Map<String, PrevFilePaths> prevFilePaths;

/**
* Resolves {@link #prevFilePaths} on demand. Resolution is deferred until the first
* {@link #getPrevFilePaths()} call, so a callback that never reads the previous paths pays
* nothing (no FileSystemView access at all). Transient because it captures a
* FileSystemView supplier, which is not serializable: on a deserialized instance this is
* null and the already-resolved {@link #prevFilePaths} is used instead. Excluded from the
* generated getters so the {@link Lazy} wrapper never leaks into JSON.
*/
@Getter(AccessLevel.NONE)
private final transient Lazy<Map<String, PrevFilePaths>> prevFilePathsResolver;

/**
* Free-form context that producers can attach for downstream callback consumers.
* The OSS write client populates this as empty; specialized callsites or wrappers
* may populate it with whatever context their callbacks need.
*/
private final Map<String, String> extraContext;

public HoodieWriteCommitCallbackMessage(String commitTime,
Comment thread
codope marked this conversation as resolved.
String tableName,
Comment thread
codope marked this conversation as resolved.
String basePath,
List<HoodieWriteStat> hoodieWriteStat,
Option<String> commitActionType,
Option<Map<String, String>> extraMetadata,
Supplier<BaseFileOnlyView> fsViewSupplier,
Map<String, String> extraContext) {
this.commitTime = commitTime;
this.tableName = tableName;
this.basePath = basePath;
this.hoodieWriteStat = hoodieWriteStat;
this.commitActionType = commitActionType;
this.extraMetadata = extraMetadata;
this.prevFilePathsResolver = Lazy.lazily(() -> HoodieWriteCommitCallbackUtil.resolvePrevFilePaths(
hoodieWriteStat, fsViewSupplier == null ? null : fsViewSupplier.get()));
this.extraContext = extraContext;
}

public HoodieWriteCommitCallbackMessage(String commitTime,
String tableName,
String basePath,
List<HoodieWriteStat> hoodieWriteStat) {
this(commitTime, tableName, basePath, hoodieWriteStat, Option.empty(), Option.empty());
this(commitTime, tableName, basePath, hoodieWriteStat, Option.empty(), Option.empty(),
null, Collections.emptyMap());
}

public HoodieWriteCommitCallbackMessage(String commitTime,
String tableName,
String basePath,
List<HoodieWriteStat> hoodieWriteStat,
Option<String> commitActionType,
Option<Map<String, String>> extraMetadata) {
this(commitTime, tableName, basePath, hoodieWriteStat, commitActionType, extraMetadata,
null, Collections.emptyMap());
}

/**
* Returns the previous base file paths keyed by fileId, resolving them from the file-system
* view on first access and memoizing the result. A consumer that never calls this triggers
* no FileSystemView lookup. Never null: empty when no view was supplied and when the commit
* only inserted.
*/
public Map<String, PrevFilePaths> getPrevFilePaths() {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sorry sagar, I asked Codex and it seems the Ser/De customization is unavoidable:

The PR description already claims this behavior, but the current implementation does not provide it: `prevFilePaths` is a `transient Lazy`, and the round-trip test explicitly expects the paths to disappear.

Use two fields:

- A normal serializable field containing the resolved map.
- A transient lazy resolver containing the non-serializable filesystem-view supplier.

Then force resolution only when Java serialization actually occurs.

```java
import java.io.IOException;
import java.io.ObjectOutputStream;

@Getter(AccessLevel.NONE)
private volatile Map<String, PrevFilePaths> prevFilePaths;

@Getter(AccessLevel.NONE)
private final transient Lazy<Map<String, PrevFilePaths>> prevFilePathsResolver;

Initialize them in the constructor:

this.prevFilePaths = null;
this.prevFilePathsResolver = Lazy.lazily(() ->
    HoodieWriteCommitCallbackUtil.resolvePrevFilePaths(
        hoodieWriteStat,
        fsViewSupplier == null ? null : fsViewSupplier.get()));

The getter memoizes the resolved value:

public Map<String, PrevFilePaths> getPrevFilePaths() {
  Map<String, PrevFilePaths> paths = prevFilePaths;
  if (paths == null) {
    Lazy<Map<String, PrevFilePaths>> resolver = prevFilePathsResolver;
    paths = resolver == null
        ? Collections.emptyMap()
        : resolver.get();

    prevFilePaths = paths == null
        ? Collections.emptyMap()
        : paths;
  }

  return prevFilePaths;
}

Finally, add the serialization hook:

private void writeObject(ObjectOutputStream out) throws IOException {
  // The resolver cannot cross the serialization boundary, so materialize
  // its value at the last possible moment.
  getPrevFilePaths();
  out.defaultWriteObject();
}

This gives you the desired lifecycle:

message construction
    ↓ no filesystem-view access
callback ignores prevFilePaths
    ↓ still no access
getPrevFilePaths() or JSON/Java serialization
    ↓ resolve once and memoize
deserialized message
    ↓ resolved map remains available; resolver is null

Do not clear prevFilePathsResolver manually after resolution. Hudi’s Lazy already clears its initializer, and keeping the Lazy reference avoids subtle concurrency races.

Update the serialization test to expect preservation:

@Test
public void javaSerializationResolvesAndPreservesPrevFilePaths()
    throws IOException, ClassNotFoundException {
  AtomicInteger lookups = new AtomicInteger();

  HoodieWriteCommitCallbackMessage message =
      new HoodieWriteCommitCallbackMessage(
          COMMIT_TIME, "table", "/base", updateStat(),
          Option.of("commit"), Option.empty(),
          () -> {
            lookups.incrementAndGet();
            return viewResolving(PREV_PATH);
          },
          Collections.emptyMap());

  assertEquals(0, lookups.get());

  HoodieWriteCommitCallbackMessage roundTripped =
      serializeAndDeserialize(message);

  assertEquals(1, lookups.get());
  assertEquals(
      PREV_PATH,
      roundTripped.getPrevFilePaths().get("f0").getBaseFilePath());
}

Keep the existing serialVersionUID. For compatibility with streams produced before this field existed, the getter’s resolver == null fallback returns an empty map.

This is the unavoidable serialization boundary: the filesystem view itself cannot be shipped, so Java serialization must materialize the lazy value. JSON callbacks already do this naturally because Jackson invokes getPrevFilePaths() while generating the payload. See the current [PR changes](https://github.com/apache/hudi/pull/18988/changes).

@codope codope Jul 31, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @danny0405. I'll take this and push the two-field version. One thing to correct though:

it seems the Ser/De customization is unavoidable. The PR description already claims this behavior, but the current implementation does not provide it...

That PR description was outdated. Nothing actually forces the behaviour. All three built-in callbacks (HTTP, Kafka, Pulsar) go through convertToJsonString, and Jackson invokes getPrevFilePaths() while generating the payload, so the JSON path resolves the paths either way. Nothing in the repo Java-serializes the message, and prevFilePaths is new in this PR, so no consumer can be depending on it yet. I had also tested after making this change.

But, i agree with your choice. The class is a @PublicAPIClass and implements Serializable, so a custom callback that ships the message across a JVM boundary would silently get an empty map rather than an error.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, then the serialization custimization might just be the API compatibility instead of real use usage here.

@codope codope Jul 31, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah that's correct. Updated now. thanks
6c5d2d0

Map<String, PrevFilePaths> paths = prevFilePaths;
if (paths == null) {
// The resolver is null only on an instance restored from Java serialization, and there
// the resolved map has already been read back into prevFilePaths (see writeObject).
paths = prevFilePathsResolver == null ? Collections.emptyMap() : prevFilePathsResolver.get();
prevFilePaths = paths;
}
return paths;
}

/**
* A {@link BaseFileOnlyView} cannot cross a serialization boundary, so materialize the
* paths at the last possible moment and let the resolved map travel in their place.
*/
private void writeObject(ObjectOutputStream out) throws IOException {
getPrevFilePaths();
out.defaultWriteObject();
}

/**
* Container for previously-existing file paths associated with a single fileId in a
* commit. {@link #baseFilePath} is the base file the new write replaces, and
* {@link #bootstrapBaseFilePath} is the bootstrap-source file the previous
* base file referenced (null for non-bootstrap tables).
*/
@Getter
public static class PrevFilePaths implements Serializable {
private static final long serialVersionUID = 1L;
private final String baseFilePath;
private final String bootstrapBaseFilePath;

public PrevFilePaths(String baseFilePath, String bootstrapBaseFilePath) {
this.baseFilePath = baseFilePath;
this.bootstrapBaseFilePath = bootstrapBaseFilePath;
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,9 @@

import org.apache.hudi.avro.model.HoodieCleanMetadata;
import org.apache.hudi.callback.HoodieClientInitCallback;
import org.apache.hudi.callback.HoodieCommitCallbackFactory;
import org.apache.hudi.callback.HoodieWriteCommitCallback;
import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage;
import org.apache.hudi.client.embedded.EmbeddedTimelineServerHelper;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient;
Expand All @@ -34,6 +37,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.HoodieStorageUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
Expand Down Expand Up @@ -63,6 +67,7 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static org.apache.hudi.config.HoodieWriteConfig.APPLICATION_ID;
Expand All @@ -87,6 +92,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
Expand Down Expand Up @@ -462,4 +475,36 @@ 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,
fsViewSupplier,
Collections.emptyMap()));
} catch (Exception e) {
log.warn("HoodieWriteCommitCallback failed for commit {} ({}); ignoring",
commitTime, commitActionType, e);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -424,6 +424,8 @@ protected void completeCompaction(HoodieCommitMetadata metadata, HoodieTable tab
);
}
log.info("Compacted successfully on commit {}", compactionCommitTime);
fireCommitCallbackIfNecessary(compactionCommitTime, HoodieTimeline.COMMIT_ACTION,
Comment thread
codope marked this conversation as resolved.
Comment thread
codope marked this conversation as resolved.
writeStats, table::getBaseFileOnlyView, Option.empty());
} finally {
if (config.getWriteConcurrencyMode().supportsMultiWriter()) {
this.heartbeatClient.stop(compactionCommitTime);
Expand Down Expand Up @@ -496,6 +498,8 @@ protected void completeLogCompaction(HoodieCommitMetadata metadata, HoodieTable
);
}
log.info("Log Compacted successfully on commit {}", logCompactionCommitTime);
fireCommitCallbackIfNecessary(logCompactionCommitTime, HoodieTimeline.DELTA_COMMIT_ACTION,
writeStats, table::getBaseFileOnlyView, Option.empty());
}

/**
Expand Down Expand Up @@ -640,6 +644,8 @@ private void completeClustering(HoodieReplaceCommitMetadata replaceCommitMetadat
heartbeatClient.stop(clusteringCommitTime);
}
log.info("Clustering successfully on commit {} for table {}", clusteringCommitTime, table.getConfig().getBasePath());
fireCommitCallbackIfNecessary(clusteringCommitTime, clusteringInstant.getAction(),
writeStats, table::getBaseFileOnlyView, Option.empty());
}

protected void runTableServicesInline(HoodieTable table, HoodieCommitMetadata metadata, Option<Map<String, String>> extraMetadata) {
Expand Down
Loading
Loading