This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new bf3cc4393981 feat(client): enrich write commit callback message and
fire it for table-service commits (#18988)
bf3cc4393981 is described below
commit bf3cc4393981149167973e700927f22748ae1dcf
Author: Sagar Sumit <[email protected]>
AuthorDate: Fri Jul 31 13:23:25 2026 +0530
feat(client): enrich write commit callback message and fire it for
table-service commits (#18988)
* feat(client): enrich write commit callback message and fire it for
table-service commits
Two backward-compatible improvements to the post-commit write callback
mechanism:
1. Enrich HoodieWriteCommitCallbackMessage with two optional fields so
callback
implementations no longer have to rebuild a FileSystemView or reach into
engine
config:
- prevFilePaths: Map<fileId, PrevFilePaths> -- the previous base file
(and
bootstrap source, if any) each updated file group replaces,
pre-resolved by the
write client from its cached file-system view.
- extraContext: Map<String,String> -- free-form context producers can
attach.
Both default to empty maps; the existing 4-arg and 6-arg constructors
are preserved.
2. Fire the callback for table-service commits too (compaction and
clustering
completion), not just data commits. The shared firing logic
(fireCommitCallback)
and prev-file resolution (resolvePrevFilePaths) are lifted into
BaseHoodieClient so
both BaseHoodieWriteClient (data commits, via postCommit) and
BaseHoodieTableServiceClient (compaction/clustering completion) reuse
them. The
commitCallback field is lifted up from BaseHoodieWriteClient.
postCommit now receives the resolved commit action type so the callback
reports the
actual action (e.g. replacecommit for insert_overwrite) rather than the
table's base
action type.
Best-effort by design: callback and prev-file resolution failures are
logged and never
fail the write.
Adds TestBaseHoodieClient covering resolvePrevFilePaths (inserts, updates,
bootstrap
capture, missing-file skip, best-effort on view failure, null inputs) and
the message
default/retention contract.
* address commetns
Signed-off-by: codope <[email protected]>
* resolve supplier comment and other nits by bot
Signed-off-by: codope <[email protected]>
* rebase and address a few minor comments
Signed-off-by: codope <[email protected]>
* address recent comment on PrevFilePaths utility and others
Signed-off-by: codope <[email protected]>
* compute prev file paths from the file system view supplier in the
callback message
Signed-off-by: codope <[email protected]>
* address serde comment
Signed-off-by: codope <[email protected]>
---------
Signed-off-by: codope <[email protected]>
---
.../callback/HoodieWriteCommitCallbackUtil.java | 54 ++++++
.../common/HoodieWriteCommitCallbackMessage.java | 118 +++++++++++-
.../org/apache/hudi/client/BaseHoodieClient.java | 45 +++++
.../hudi/client/BaseHoodieTableServiceClient.java | 6 +
.../apache/hudi/client/BaseHoodieWriteClient.java | 27 ++-
.../TestHoodieWriteCommitCallbackUtil.java | 146 +++++++++++++++
.../TestHoodieWriteCommitCallbackMessage.java | 198 +++++++++++++++++++++
.../hudi/client/HoodieFlinkTableServiceClient.java | 4 +
.../TestHoodieJavaClientOnMergeOnReadStorage.java | 121 +++++++++++++
9 files changed, 700 insertions(+), 19 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/HoodieWriteCommitCallbackUtil.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/HoodieWriteCommitCallbackUtil.java
index 6ae37897af40..6f4ee4b17eeb 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/HoodieWriteCommitCallbackUtil.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/HoodieWriteCommitCallbackUtil.java
@@ -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();
@@ -41,4 +53,46 @@ public class HoodieWriteCommitCallbackUtil {
}
}
+ /**
+ * 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;
+ }
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/common/HoodieWriteCommitCallbackMessage.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/common/HoodieWriteCommitCallbackMessage.java
index 713427b52c01..30c47d011ea8 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/common/HoodieWriteCommitCallbackMessage.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/common/HoodieWriteCommitCallbackMessage.java
@@ -19,20 +19,26 @@ package org.apache.hudi.callback.common;
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 {
@@ -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,
+ String tableName,
+ 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() {
+ 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;
+ }
}
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java
index 379c849d6a9f..444ffd45e96b 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java
@@ -20,6 +20,9 @@ package org.apache.hudi.client;
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;
@@ -34,6 +37,7 @@ import org.apache.hudi.common.table.timeline.HoodieTimeline;
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;
@@ -63,6 +67,7 @@ import java.util.HashSet;
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;
@@ -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
@@ -462,4 +475,36 @@ public abstract class BaseHoodieClient implements
Serializable, AutoCloseable {
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);
+ }
+ }
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
index 24ace4f86180..235b1023cdae 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
@@ -424,6 +424,8 @@ public abstract class BaseHoodieTableServiceClient<I, T, O>
extends BaseHoodieCl
);
}
log.info("Compacted successfully on commit {}", compactionCommitTime);
+ fireCommitCallbackIfNecessary(compactionCommitTime,
HoodieTimeline.COMMIT_ACTION,
+ writeStats, table::getBaseFileOnlyView, Option.empty());
} finally {
if (config.getWriteConcurrencyMode().supportsMultiWriter()) {
this.heartbeatClient.stop(compactionCommitTime);
@@ -496,6 +498,8 @@ public abstract class BaseHoodieTableServiceClient<I, T, O>
extends BaseHoodieCl
);
}
log.info("Log Compacted successfully on commit {}",
logCompactionCommitTime);
+ fireCommitCallbackIfNecessary(logCompactionCommitTime,
HoodieTimeline.DELTA_COMMIT_ACTION,
+ writeStats, table::getBaseFileOnlyView, Option.empty());
}
/**
@@ -640,6 +644,8 @@ public abstract class BaseHoodieTableServiceClient<I, T, O>
extends BaseHoodieCl
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) {
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index c43c9bcaa070..4e0e267391c5 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -24,9 +24,6 @@ import org.apache.hudi.avro.model.HoodieIndexPlan;
import org.apache.hudi.avro.model.HoodieRestoreMetadata;
import org.apache.hudi.avro.model.HoodieRestorePlan;
import org.apache.hudi.avro.model.HoodieRollbackMetadata;
-import org.apache.hudi.callback.HoodieCommitCallbackFactory;
-import org.apache.hudi.callback.HoodieWriteCommitCallback;
-import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage;
import org.apache.hudi.callback.common.WriteStatusValidator;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
import org.apache.hudi.client.heartbeat.WriterHeartbeatUtils;
@@ -149,7 +146,6 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
@Getter
@Setter
private transient WriteOperationType operationType;
- private transient HoodieWriteCommitCallback commitCallback;
protected transient Timer.Context writeTimer = null;
@@ -290,7 +286,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
boolean postCommitStatus = true;
HoodieTimer postCommitTimer = HoodieTimer.start();
try {
- postCommit(table, metadata, instantTime, extraMetadata);
+ postCommit(table, metadata, instantTime, commitActionType,
extraMetadata);
mayBeCleanAndArchive(table);
runTableServicesInline(table, metadata, extraMetadata);
} catch (Exception e) {
@@ -306,15 +302,6 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
}
emitCommitMetrics(instantTime, metadata, commitActionType);
-
- // callback if needed.
- if (config.writeCommitCallbackOn()) {
- if (null == commitCallback) {
- commitCallback = HoodieCommitCallbackFactory.create(config);
- }
- commitCallback.call(new HoodieWriteCommitCallbackMessage(
- instantTime, config.getTableName(), config.getBasePath(),
tableWriteStats.getDataTableWriteStats(), Option.of(commitActionType),
extraMetadata));
- }
return true;
}
@@ -645,7 +632,9 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
boolean postCommitStatus = true;
HoodieTimer postCommitTimer = HoodieTimer.start();
try {
- postCommit(hoodieTable, result.getCommitMetadata().get(), instantTime,
Option.empty());
+ String commitActionType =
CommitUtils.getCommitActionType(operationType,
hoodieTable.getMetaClient().getTableType());
+ postCommit(hoodieTable, result.getCommitMetadata().get(), instantTime,
+ commitActionType, Option.empty());
mayBeCleanAndArchive(hoodieTable);
} catch (Exception e) {
postCommitStatus = false;
@@ -672,7 +661,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
* @param instantTime Instant Time
* @param extraMetadata Additional Metadata passed by user
*/
- protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata,
String instantTime, Option<Map<String, String>> extraMetadata) {
+ protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata,
String instantTime, String commitActionType, Option<Map<String, String>>
extraMetadata) {
try {
context.setJobStatus(this.getClass().getSimpleName(), "Cleaning up
marker directories for commit " + instantTime + " in table "
+ config.getTableName());
@@ -680,6 +669,12 @@ public abstract class BaseHoodieWriteClient<T, I, K, O>
extends BaseHoodieClient
WriteMarkersFactory.get(config.getMarkersType(), table, instantTime)
.quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism());
metrics.updateTableServiceInstantMetrics(table.getActiveTimeline());
+ // Fire write commit callback if a callback class is registered.
postCommit() is reached
+ // by both auto-commit and explicit-commit paths; compaction and
clustering have their own
+ // explicit fireCommitCallbackIfNecessary call sites in
BaseHoodieTableServiceClient.
+ List<HoodieWriteStat> stats = metadata.getWriteStats();
+ fireCommitCallbackIfNecessary(instantTime, commitActionType, stats,
+ table::getBaseFileOnlyView, extraMetadata);
} finally {
this.heartbeatClient.stop(instantTime);
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/callback/TestHoodieWriteCommitCallbackUtil.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/callback/TestHoodieWriteCommitCallbackUtil.java
new file mode 100644
index 000000000000..7bb1d518451e
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/callback/TestHoodieWriteCommitCallbackUtil.java
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for
+ * {@link HoodieWriteCommitCallbackUtil#resolvePrevFilePaths(List,
BaseFileOnlyView)}, which
+ * pre-resolves the previous base file (and bootstrap source, if any) for each
updated file
+ * group from a cached {@link BaseFileOnlyView}, so callback implementations
receive the
+ * read/write file pairing without rebuilding a file-system view.
+ */
+public class TestHoodieWriteCommitCallbackUtil {
+
+ private static final String PARTITION = "2024/01/01";
+ private static final String PREV_COMMIT = "001";
+
+ private static HoodieWriteStat stat(String fileId, String partitionPath,
String prevCommit) {
+ HoodieWriteStat writeStat = new HoodieWriteStat();
+ writeStat.setFileId(fileId);
+ writeStat.setPartitionPath(partitionPath);
+ writeStat.setPrevCommit(prevCommit);
+ return writeStat;
+ }
+
+ @Test
+ public void resolvePrevFilePathsReturnsEmptyForNullInputs() {
+ BaseFileOnlyView view = mock(BaseFileOnlyView.class);
+ assertTrue(HoodieWriteCommitCallbackUtil.resolvePrevFilePaths(null,
view).isEmpty(),
+ "null stats must yield an empty map");
+ assertTrue(HoodieWriteCommitCallbackUtil.resolvePrevFilePaths(
+ Collections.singletonList(stat("f0", PARTITION, PREV_COMMIT)),
null).isEmpty(),
+ "null file-system view must yield an empty map");
+ }
+
+ @Test
+ public void resolvePrevFilePathsSkipsStatsWithoutAPrevCommit() {
+ BaseFileOnlyView view = mock(BaseFileOnlyView.class);
+ List<HoodieWriteStat> inserts = Arrays.asList(
+ stat("f-null", PARTITION, null),
+ stat("f-empty", PARTITION, ""),
+ stat("f-nullcommit", PARTITION, HoodieWriteStat.NULL_COMMIT));
+
+ Map<String, PrevFilePaths> resolved =
+ HoodieWriteCommitCallbackUtil.resolvePrevFilePaths(inserts, view);
+
+ assertTrue(resolved.isEmpty(), "inserts (no prevCommit) must not resolve a
prev base file");
+ // The view must not even be consulted for inserts.
+ verify(view, never()).getBaseFileOn(anyString(), anyString(), anyString());
+ }
+
+ @Test
+ public void resolvePrevFilePathsResolvesUpdatePrevBaseFile() {
+ BaseFileOnlyView view = mock(BaseFileOnlyView.class);
+ HoodieBaseFile prevBase = new HoodieBaseFile("/tbl/" + PARTITION +
"/f0_0-1-1_" + PREV_COMMIT + ".parquet");
+ when(view.getBaseFileOn(PARTITION, PREV_COMMIT,
"f0")).thenReturn(Option.of(prevBase));
+
+ Map<String, PrevFilePaths> resolved =
HoodieWriteCommitCallbackUtil.resolvePrevFilePaths(
+ Collections.singletonList(stat("f0", PARTITION, PREV_COMMIT)), view);
+
+ assertEquals(1, resolved.size());
+ assertEquals(prevBase.getPath(), resolved.get("f0").getBaseFilePath());
+ assertNull(resolved.get("f0").getBootstrapBaseFilePath(), "non-bootstrap
update has no bootstrap path");
+ }
+
+ @Test
+ public void resolvePrevFilePathsCapturesBootstrapBaseFile() {
+ BaseFileOnlyView view = mock(BaseFileOnlyView.class);
+ BaseFile bootstrap = new BaseFile("/bootstrap/source/f0.parquet");
+ HoodieBaseFile prevBase = new HoodieBaseFile("/tbl/" + PARTITION +
"/f0_0-1-1_" + PREV_COMMIT + ".parquet", bootstrap);
+ when(view.getBaseFileOn(PARTITION, PREV_COMMIT,
"f0")).thenReturn(Option.of(prevBase));
+
+ Map<String, PrevFilePaths> resolved =
HoodieWriteCommitCallbackUtil.resolvePrevFilePaths(
+ Collections.singletonList(stat("f0", PARTITION, PREV_COMMIT)), view);
+
+ assertEquals(prevBase.getPath(), resolved.get("f0").getBaseFilePath());
+ assertEquals(bootstrap.getPath(),
resolved.get("f0").getBootstrapBaseFilePath(),
+ "bootstrap source path must be carried through for bootstrapped file
groups");
+ }
+
+ @Test
+ public void resolvePrevFilePathsSkipsWhenBaseFileAbsent() {
+ BaseFileOnlyView view = mock(BaseFileOnlyView.class);
+ when(view.getBaseFileOn(PARTITION, PREV_COMMIT,
"f0")).thenReturn(Option.empty());
+
+ Map<String, PrevFilePaths> resolved =
HoodieWriteCommitCallbackUtil.resolvePrevFilePaths(
+ Collections.singletonList(stat("f0", PARTITION, PREV_COMMIT)), view);
+
+ assertTrue(resolved.isEmpty(), "a missing prev base file must be skipped,
not mapped to null");
+ }
+
+ @Test
+ public void resolvePrevFilePathsIsBestEffortOnViewFailure() {
+ BaseFileOnlyView view = mock(BaseFileOnlyView.class);
+ when(view.getBaseFileOn(PARTITION, PREV_COMMIT, "boom"))
+ .thenThrow(new RuntimeException("stale or remote view error"));
+ HoodieBaseFile prevBase = new HoodieBaseFile("/tbl/" + PARTITION +
"/ok_0-1-1_" + PREV_COMMIT + ".parquet");
+ when(view.getBaseFileOn(PARTITION, PREV_COMMIT,
"ok")).thenReturn(Option.of(prevBase));
+
+ Map<String, PrevFilePaths> resolved =
HoodieWriteCommitCallbackUtil.resolvePrevFilePaths(
+ Arrays.asList(stat("boom", PARTITION, PREV_COMMIT), stat("ok",
PARTITION, PREV_COMMIT)), view);
+
+ // The failing file group is dropped; resolution continues for the rest
(must not fail the commit).
+ assertFalse(resolved.containsKey("boom"));
+ assertEquals(prevBase.getPath(), resolved.get("ok").getBaseFilePath());
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/callback/common/TestHoodieWriteCommitCallbackMessage.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/callback/common/TestHoodieWriteCommitCallbackMessage.java
new file mode 100644
index 000000000000..5313930561a7
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/callback/common/TestHoodieWriteCommitCallbackMessage.java
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.callback.common;
+
+import org.apache.hudi.callback.HoodieWriteCommitCallbackUtil;
+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.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for the {@link HoodieWriteCommitCallbackMessage} contract:
default (never-null)
+ * collections, lazy one-shot resolution of {@code prevFilePaths} off the
supplied file-system
+ * view, and the Java-serialization round trip (the view cannot be shipped, so
the resolved
+ * paths must be materialized at the boundary and carried across in its place).
+ */
+public class TestHoodieWriteCommitCallbackMessage {
+
+ private static final String COMMIT_TIME = "002";
+ private static final String PARTITION = "2024/01/01";
+ private static final String PREV_COMMIT = "001";
+ private static final String PREV_PATH = "/tbl/" + PARTITION + "/f0_0-1-1_" +
PREV_COMMIT + ".parquet";
+ private static final String BOOTSTRAP_PATH = "/bootstrap/source/f0.parquet";
+
+ private static List<HoodieWriteStat> updateStat() {
+ HoodieWriteStat writeStat = new HoodieWriteStat();
+ writeStat.setFileId("f0");
+ writeStat.setPartitionPath(PARTITION);
+ writeStat.setPrevCommit(PREV_COMMIT);
+ return Collections.singletonList(writeStat);
+ }
+
+ private static BaseFileOnlyView viewResolving(String prevBaseFilePath) {
+ BaseFileOnlyView view = mock(BaseFileOnlyView.class);
+ when(view.getBaseFileOn(PARTITION, PREV_COMMIT, "f0"))
+ .thenReturn(Option.of(new HoodieBaseFile(prevBaseFilePath)));
+ return view;
+ }
+
+ private static BaseFileOnlyView viewResolvingWithBootstrap(String
prevBaseFilePath, String bootstrapPath) {
+ BaseFileOnlyView view = mock(BaseFileOnlyView.class);
+ when(view.getBaseFileOn(PARTITION, PREV_COMMIT, "f0"))
+ .thenReturn(Option.of(new HoodieBaseFile(prevBaseFilePath, new
BaseFile(bootstrapPath))));
+ return view;
+ }
+
+ @Test
+ public void callbackMessageDefaultsCollectionsToEmpty() {
+ HoodieWriteCommitCallbackMessage message = new
HoodieWriteCommitCallbackMessage(
+ COMMIT_TIME, "table", "/base", Collections.emptyList());
+
+ assertFalse(message.getCommitActionType().isPresent());
+ assertFalse(message.getExtraMetadata().isPresent());
+ assertTrue(message.getPrevFilePaths().isEmpty(), "prevFilePaths must
default to an empty map, never null");
+ assertTrue(message.getExtraContext().isEmpty(), "extraContext must default
to an empty map, never null");
+ }
+
+ @Test
+ public void callbackMessageResolvesPrevFilePathsFromViewAndRetainsContext() {
+ Map<String, String> extraContext = Collections.singletonMap("file_id",
"f0");
+
+ HoodieWriteCommitCallbackMessage message = new
HoodieWriteCommitCallbackMessage(
+ COMMIT_TIME, "table", "/base", updateStat(),
+ Option.of("commit"), Option.empty(), () -> viewResolving(PREV_PATH),
extraContext);
+
+ assertEquals("commit", message.getCommitActionType().get());
+ PrevFilePaths resolved = message.getPrevFilePaths().get("f0");
+ assertEquals(PREV_PATH, resolved.getBaseFilePath());
+ assertEquals(extraContext, message.getExtraContext());
+ }
+
+ @Test
+ public void nullFileSystemViewSupplierYieldsEmptyPrevFilePaths() {
+ HoodieWriteCommitCallbackMessage message = new
HoodieWriteCommitCallbackMessage(
+ COMMIT_TIME, "table", "/base", updateStat(),
+ Option.of("commit"), Option.empty(), null, Collections.emptyMap());
+
+ assertTrue(message.getPrevFilePaths().isEmpty(),
+ "a message built without a file-system view must yield an empty map,
never null");
+ }
+
+ @Test
+ public void prevFilePathsAreResolvedLazilyAndMemoized() {
+ AtomicInteger viewLookups = new AtomicInteger();
+ BaseFileOnlyView view = viewResolving(PREV_PATH);
+ Supplier<BaseFileOnlyView> viewSupplier = () -> {
+ viewLookups.incrementAndGet();
+ return view;
+ };
+
+ HoodieWriteCommitCallbackMessage message = new
HoodieWriteCommitCallbackMessage(
+ COMMIT_TIME, "table", "/base", updateStat(),
+ Option.empty(), Option.empty(), viewSupplier, Collections.emptyMap());
+
+ // Constructing the message must not touch the file-system view.
+ assertEquals(0, viewLookups.get(),
+ "prevFilePaths must not be resolved until a consumer reads them");
+ verify(view, never()).getBaseFileOn(anyString(), anyString(), anyString());
+
+ assertEquals(PREV_PATH,
message.getPrevFilePaths().get("f0").getBaseFilePath());
+ // A second read must reuse the memoized result rather than resolve again.
+ assertEquals(PREV_PATH,
message.getPrevFilePaths().get("f0").getBaseFilePath());
+ assertEquals(1, viewLookups.get(), "prevFilePaths must be resolved at most
once and memoized");
+ verify(view).getBaseFileOn(PARTITION, PREV_COMMIT, "f0");
+ }
+
+ @Test
+ public void javaSerializationResolvesAndPreservesPrevFilePaths() throws
IOException, ClassNotFoundException {
+ AtomicInteger viewLookups = new AtomicInteger();
+ // The view supplier cannot be shipped, so writeObject has to materialize
the paths first.
+ HoodieWriteCommitCallbackMessage message = new
HoodieWriteCommitCallbackMessage(
+ COMMIT_TIME, "table", "/base", updateStat(),
+ Option.of("commit"), Option.empty(),
+ () -> {
+ viewLookups.incrementAndGet();
+ return viewResolvingWithBootstrap(PREV_PATH, BOOTSTRAP_PATH);
+ },
+ Collections.emptyMap());
+
+ assertEquals(0, viewLookups.get(), "building the message must not touch
the file-system view");
+
+ HoodieWriteCommitCallbackMessage roundTripped =
serializeAndDeserialize(message);
+
+ assertEquals(1, viewLookups.get(), "serialization must force resolution
exactly once");
+ assertEquals(COMMIT_TIME, roundTripped.getCommitTime());
+ assertEquals("commit", roundTripped.getCommitActionType().get());
+ assertEquals(1, roundTripped.getHoodieWriteStat().size());
+ assertEquals(PREV_PATH,
roundTripped.getPrevFilePaths().get("f0").getBaseFilePath());
+ assertEquals(BOOTSTRAP_PATH,
roundTripped.getPrevFilePaths().get("f0").getBootstrapBaseFilePath(),
+ "the bootstrap source path must survive the round trip too");
+ }
+
+ @Test
+ public void jsonPayloadExposesPrevFilePathsAndNotTheResolver() {
+ HoodieWriteCommitCallbackMessage message = new
HoodieWriteCommitCallbackMessage(
+ COMMIT_TIME, "table", "/base", updateStat(),
+ Option.of("commit"), Option.empty(), () -> viewResolving(PREV_PATH),
Collections.emptyMap());
+
+ // This is the payload the built-in HTTP/Kafka/Pulsar callbacks put on the
wire.
+ String json = HoodieWriteCommitCallbackUtil.convertToJsonString(message);
+
+ assertTrue(json.contains("\"prevFilePaths\""), "prevFilePaths must be part
of the callback payload");
+ assertTrue(json.contains(PREV_PATH), "Jackson must see the resolved paths,
not the lazy holder");
+ assertFalse(json.contains("prevFilePathsResolver"),
+ "the lazy resolver is an implementation detail and must never reach
the payload");
+ }
+
+ private static HoodieWriteCommitCallbackMessage serializeAndDeserialize(
+ HoodieWriteCommitCallbackMessage message) throws IOException,
ClassNotFoundException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (ObjectOutputStream out = new ObjectOutputStream(bytes)) {
+ out.writeObject(message);
+ }
+ try (ObjectInputStream in = new ObjectInputStream(new
ByteArrayInputStream(bytes.toByteArray()))) {
+ return (HoodieWriteCommitCallbackMessage) in.readObject();
+ }
+ }
+}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
index 929515115627..0ab62d498ddd 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
@@ -100,6 +100,8 @@ public class HoodieFlinkTableServiceClient<T> extends
BaseHoodieTableServiceClie
}
}
log.info("Compacted successfully on commit {}", compactionCommitTime);
+ fireCommitCallbackIfNecessary(compactionCommitTime,
HoodieActiveTimeline.COMMIT_ACTION,
+ metadata.getWriteStats(), table::getBaseFileOnlyView,
Option.empty());
} finally {
if (config.getWriteConcurrencyMode().supportsMultiWriter()) {
this.heartbeatClient.stop(compactionCommitTime);
@@ -160,6 +162,8 @@ public class HoodieFlinkTableServiceClient<T> extends
BaseHoodieTableServiceClie
}
}
log.info("Clustering successfully on commit {}", clusteringCommitTime);
+ fireCommitCallbackIfNecessary(clusteringCommitTime,
clusteringInstant.getAction(),
+ writeStats, table::getBaseFileOnlyView, Option.empty());
}
@Override
diff --git
a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnMergeOnReadStorage.java
b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnMergeOnReadStorage.java
index 8f4a73f51b08..3964dd69a517 100644
---
a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnMergeOnReadStorage.java
+++
b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnMergeOnReadStorage.java
@@ -18,14 +18,21 @@
package org.apache.hudi.client.functional;
+import org.apache.hudi.callback.HoodieWriteCommitCallback;
+import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage;
import org.apache.hudi.client.HoodieJavaWriteClient;
import org.apache.hudi.client.WriteClientTestUtils;
+import
org.apache.hudi.client.clustering.plan.strategy.JavaSizeBasedClusteringPlanStrategy;
+import
org.apache.hudi.client.clustering.run.strategy.JavaSortAndSizeExecutionStrategy;
import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.view.SyncableFileSystemView;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.testutils.HoodieTestTable;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieClusteringConfig;
import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.config.HoodieWriteCommitCallbackConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.table.action.HoodieWriteMetadata;
@@ -37,12 +44,16 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.stream.Collectors;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.TIMELINE_FACTORY;
import static
org.apache.hudi.testutils.GenericRecordValidationTestUtils.assertDataInMORTable;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestHoodieJavaClientOnMergeOnReadStorage extends
HoodieJavaClientTestHarness {
@@ -180,4 +191,114 @@ public class TestHoodieJavaClientOnMergeOnReadStorage
extends HoodieJavaClientTe
return HoodieTableType.MERGE_ON_READ;
}
+ @Test
+ public void testWriteCommitCallbackFiresOnCompaction() throws Exception {
+ RecordingCommitCallback.MESSAGES.clear();
+ HoodieWriteConfig config =
getConfigBuilder(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA,
+ HoodieIndex.IndexType.INMEMORY)
+
.withCompactionConfig(HoodieCompactionConfig.newBuilder().withMaxNumDeltaCommitsBeforeCompaction(2).build())
+ .withCallbackConfig(HoodieWriteCommitCallbackConfig.newBuilder()
+ .writeCommitCallbackOn("true")
+ .withCallbackClass(RecordingCommitCallback.class.getName())
+ .build())
+ .build();
+ HoodieJavaWriteClient client = getHoodieWriteClient(config);
+
+ // Two delta commits through the auto-commit path.
+ String commitTime = WriteClientTestUtils.createNewInstantTime();
+ insertBatch(config, client, commitTime, "000", 100,
HoodieJavaWriteClient::insert,
+ false, false, 100, 100, 1, Option.empty(), INSTANT_GENERATOR);
+ String prevCommit = commitTime;
+ commitTime = WriteClientTestUtils.createNewInstantTime();
+ updateBatch(config, client, commitTime, prevCommit,
+ Option.of(Arrays.asList(prevCommit)), "000", 50,
HoodieJavaWriteClient::upsert,
+ false, false, 5, 100, 2, config.populateMetaFields(),
INSTANT_GENERATOR);
+
+ // The callback must fire for the auto-committed delta commits with the
deltacommit action.
+ assertTrue(RecordingCommitCallback.MESSAGES.stream().anyMatch(m ->
+
HoodieTimeline.DELTA_COMMIT_ACTION.equals(m.getCommitActionType().orElse(null))),
+ "callback must fire for delta commits");
+
+ // Schedule, execute and commit compaction.
+ Option<String> compactionTime = client.scheduleCompaction(Option.empty());
+ assertTrue(compactionTime.isPresent());
+ HoodieWriteMetadata writeMetadata = client.compact(compactionTime.get());
+ client.commitCompaction(compactionTime.get(), writeMetadata,
Option.empty());
+
assertTrue(metaClient.reloadActiveTimeline().filterCompletedInstants().containsInstant(compactionTime.get()));
+
+ // The callback must fire exactly once for the compaction completion,
reporting the completed
+ // timeline action (commit).
+ List<HoodieWriteCommitCallbackMessage> compactionMessages =
RecordingCommitCallback.MESSAGES.stream()
+ .filter(m -> m.getCommitTime().equals(compactionTime.get()))
+ .collect(Collectors.toList());
+ assertEquals(1, compactionMessages.size(), "callback must fire once for
the compaction commit");
+ assertEquals(HoodieTimeline.COMMIT_ACTION,
compactionMessages.get(0).getCommitActionType().orElse(null));
+ assertNotNull(compactionMessages.get(0).getPrevFilePaths(), "prevFilePaths
must never be null");
+ }
+
+ @Test
+ public void testWriteCommitCallbackFiresOnClustering() throws Exception {
+ RecordingCommitCallback.MESSAGES.clear();
+ HoodieClusteringConfig clusteringConfig =
HoodieClusteringConfig.newBuilder()
+ .withClusteringMaxNumGroups(10)
+ .withClusteringSortColumns("_row_key")
+ .withClusteringTargetPartitions(0)
+
.withClusteringPlanStrategyClass(JavaSizeBasedClusteringPlanStrategy.class.getName())
+
.withClusteringExecutionStrategyClass(JavaSortAndSizeExecutionStrategy.class.getName())
+ .build();
+ HoodieWriteConfig config =
getConfigBuilder(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA,
+ HoodieIndex.IndexType.INMEMORY)
+ .withClusteringConfig(clusteringConfig)
+ .withCallbackConfig(HoodieWriteCommitCallbackConfig.newBuilder()
+ .writeCommitCallbackOn("true")
+ .withCallbackClass(RecordingCommitCallback.class.getName())
+ .build())
+ .build();
+ HoodieJavaWriteClient client = getHoodieWriteClient(config);
+
+ // Two inserts create base-file groups that clustering can rewrite.
+ String commitTime = WriteClientTestUtils.createNewInstantTime();
+ insertBatch(config, client, commitTime, "000", 100,
HoodieJavaWriteClient::insert,
+ false, false, 100, 100, 1, Option.empty(), INSTANT_GENERATOR);
+ commitTime = WriteClientTestUtils.createNewInstantTime();
+ insertBatch(config, client, commitTime, "001", 100,
HoodieJavaWriteClient::insert,
+ false, false, 100, 200, 2, Option.empty(), INSTANT_GENERATOR);
+
+ // Schedule and execute clustering inline (shouldComplete = true completes
the commit).
+ Option<String> clusteringTime = client.scheduleClustering(Option.empty());
+ assertTrue(clusteringTime.isPresent(), "expected a clustering plan to be
scheduled");
+ client.cluster(clusteringTime.get(), true);
+
assertTrue(metaClient.reloadActiveTimeline().filterCompletedInstants().containsInstant(clusteringTime.get()));
+
+ // The callback must fire once for the clustering completion, reporting
the action actually on
+ // the timeline (replacecommit for table version < 8, clustering for 8+).
+ List<HoodieWriteCommitCallbackMessage> clusteringMessages =
RecordingCommitCallback.MESSAGES.stream()
+ .filter(m -> m.getCommitTime().equals(clusteringTime.get()))
+ .collect(Collectors.toList());
+ assertEquals(1, clusteringMessages.size(), "callback must fire once for
the clustering commit");
+ String action =
clusteringMessages.get(0).getCommitActionType().orElse(null);
+ assertTrue(HoodieTimeline.REPLACE_COMMIT_ACTION.equals(action) ||
HoodieTimeline.CLUSTERING_ACTION.equals(action),
+ "clustering callback must report the timeline action, got: " + action);
+ }
+
+ /**
+ * A recording {@link HoodieWriteCommitCallback} that captures every fired
message so tests can
+ * assert the callback fires for table-service (compaction/clustering)
commits with the expected
+ * action type. Loaded reflectively from the write config, so it needs a
public
+ * {@code (HoodieWriteConfig)} constructor.
+ */
+ public static class RecordingCommitCallback implements
HoodieWriteCommitCallback {
+
+ static final List<HoodieWriteCommitCallbackMessage> MESSAGES = new
CopyOnWriteArrayList<>();
+
+ public RecordingCommitCallback(HoodieWriteConfig config) {
+ // config arg required for reflective instantiation
+ }
+
+ @Override
+ public void call(HoodieWriteCommitCallbackMessage callbackMessage) {
+ MESSAGES.add(callbackMessage);
+ }
+ }
+
}
