> 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.
+ *
+ * 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 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> 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 extraContext;
+
+ public HoodieWriteCommitCallbackMessage(String commitTime,
+ String tableName,
+ String basePath,
+ List hoodieWriteStat,
+ Option commitActionType,
+ Option> extraMetadata,
+ Supplier fsViewSupplier,
+ Map 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) {
- 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,
+ Option commitActionType,
+ Option> 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 getPrevFilePaths() {
+ Map 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/callback/util/HoodieWriteCommitCallbackUtil.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/util/HoodieWriteCommitCallbackUtil.java
index cd05b78dfcf2b..c255e31b9fc8e 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/util/HoodieWriteCommitCallbackUtil.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/callback/util/HoodieWriteCommitCallbackUtil.java
@@ -17,15 +17,27 @@
package org.apache.hudi.callback.util;
+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 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.
+ *
+ * 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 resolvePrevFilePaths(List stats,
+ BaseFileOnlyView fsView) {
+ Map 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 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 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/client/BaseHoodieClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java
index 842dc38177a59..fbd0d5e4f5794 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
@@ -18,7 +18,11 @@
package org.apache.hudi.client;
+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.util.HoodieCommitCallbackFactory;
import org.apache.hudi.client.embedded.EmbeddedTimelineServerHelper;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient;
@@ -33,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.Option;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.common.util.StringUtils;
@@ -58,11 +63,13 @@
import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
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;
/**
@@ -85,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
@@ -313,19 +328,24 @@ protected boolean isStreamingWriteToMetadataEnabled(HoodieTable table) {
}
/**
- * Merges rolling metadata from recent completed commits into the current commit metadata.
+ * Merges rolling metadata from recent completed instants into the current commit metadata.
* This method MUST be called within the transaction lock after conflict resolution.
*
* Rolling metadata keys configured via {@link HoodieWriteConfig#ROLLING_METADATA_KEYS} will be
- * automatically carried forward from recent commits. The system walks back up to
- * {@link HoodieWriteConfig#ROLLING_METADATA_TIMELINE_LOOKBACK_COMMITS} commits to find the most
- * recent value for each key. This ensures that important metadata like checkpoint information
- * remains accessible without worrying about archival or missing keys in individual commits.
+ * automatically carried forward from recent instants. The system walks back through completed
+ * commits and clean instants (in reverse completion-time order) up to
+ * {@link HoodieWriteConfig#ROLLING_METADATA_TIMELINE_LOOKBACK_COMMITS} to find the most
+ * recent value for each key.
*
* @param table HoodieTable instance (may have refreshed timeline after conflict resolution)
* @param metadata Current commit metadata to be augmented with rolling metadata
*/
protected void mergeRollingMetadata(HoodieTable table, HoodieCommitMetadata metadata) {
+ // IMPORTANT: We're inside the lock here. The timeline in 'table' is either:
+ // 1. Fresh from createTable() if no conflict resolution happened
+ // 2. Reloaded during resolveWriteConflict() if conflicts were checked
+ // In both cases, we have the latest view of the timeline.
+
// Skip for metadata table - rolling metadata is only for data tables
if (table.isMetadataTable()) {
return;
@@ -336,88 +356,155 @@ protected void mergeRollingMetadata(HoodieTable table, HoodieCommitMetadata meta
return; // No rolling metadata configured
}
- // IMPORTANT: We're inside the lock here. The timeline in 'table' is either:
- // 1. Fresh from createTable() if no conflict resolution happened
- // 2. Reloaded during resolveWriteConflict() if conflicts were checked
- // In both cases, we have the latest view of the timeline.
+ Map foundRollingMetadata = collectRollingMetadataFromTimeline(table, config, rollingKeys, metadata.getExtraMetadata());
+ for (Map.Entry entry : foundRollingMetadata.entrySet()) {
+ metadata.addMetadata(entry.getKey(), entry.getValue());
+ }
+ }
- HoodieTimeline commitsTimeline = table.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ /**
+ * Overload of {@link #mergeRollingMetadata(HoodieTable, HoodieCommitMetadata)} for clean
+ * commits. Populates {@link HoodieCleanMetadata#getExtraMetadata()} with rolling metadata
+ * values found on the active timeline.
+ *
+ * This is {@code public static} so that {@code CleanActionExecutor} (which does not extend
+ * {@code BaseHoodieClient}) can invoke it.
+ */
+ public static void mergeRollingMetadata(HoodieTable table, HoodieWriteConfig config, HoodieCleanMetadata metadata) {
+ if (table.isMetadataTable()) {
+ return;
+ }
+ Set rollingKeys = config.getRollingMetadataKeys();
+ if (rollingKeys.isEmpty()) {
+ return;
+ }
- if (commitsTimeline.empty()) {
- log.info("No previous commits found. Rolling metadata will start with current commit.");
- return; // First commit - nothing to roll forward
+ Map existing = metadata.getExtraMetadata() != null
+ ? metadata.getExtraMetadata() : Collections.emptyMap();
+ Map foundRollingMetadata = collectRollingMetadataFromTimeline(table, config, rollingKeys, existing);
+ if (!foundRollingMetadata.isEmpty()) {
+ Map merged = new HashMap<>(existing);
+ merged.putAll(foundRollingMetadata);
+ metadata.setExtraMetadata(merged);
}
+ }
- try {
- Map existingExtraMetadata = metadata.getExtraMetadata();
- Map foundRollingMetadata = new HashMap<>();
- Set remainingKeys = new HashSet<>(rollingKeys);
-
- // Remove keys that are already present with non-empty values in current commit (current values take precedence)
- for (String key : rollingKeys) {
- if (existingExtraMetadata.containsKey(key) && !StringUtils.isNullOrEmpty(existingExtraMetadata.get(key))) {
- remainingKeys.remove(key);
- }
- }
+ /**
+ * Walks backwards through completed instants (commits, replace-commits, delta-commits, and
+ * clean) on the active timeline, extracting extra-metadata values for the requested rolling
+ * keys. For commit-type instants the values come from {@link HoodieCommitMetadata#getMetadata};
+ * for clean instants they come from {@link HoodieCleanMetadata#getExtraMetadata()}.
+ *
+ * Keys already present with a non-empty value in {@code existingExtra} are skipped (empty
+ * strings are treated as "missing").
+ */
+ private static Map collectRollingMetadataFromTimeline(
+ HoodieTable table, HoodieWriteConfig config,
+ Set rollingKeys, Map existingExtra) {
- if (remainingKeys.isEmpty()) {
- log.debug("All rolling metadata keys are present in current commit. No walkback needed.");
- return;
- }
+ Map foundRollingMetadata = new HashMap<>();
+ Set remaining = new HashSet<>(rollingKeys);
- int lookbackLimit = config.getRollingMetadataTimelineLookbackCommits();
- int commitsWalkedBack = 0;
+ for (String key : rollingKeys) {
+ if (existingExtra.containsKey(key) && !StringUtils.isNullOrEmpty(existingExtra.get(key))) {
+ remaining.remove(key);
+ }
+ }
+ if (remaining.isEmpty()) {
+ log.debug("All rolling metadata keys already present. No walkback needed.");
+ return foundRollingMetadata;
+ }
- // Walk back through the timeline in reverse order (most recent first) to find values for all remaining keys
- List recentCommits = commitsTimeline.getReverseOrderedInstantsByCompletionTime()
- .limit(lookbackLimit)
- .collect(Collectors.toList());
+ int lookbackLimit = config.getRollingMetadataTimelineLookbackCommits();
+ HoodieTimeline completed = table.getActiveTimeline().filterCompletedInstants();
+ List instants = completed.getReverseOrderedInstantsByCompletionTime()
+ .filter(i -> HoodieTimeline.VALID_ACTIONS_FOR_ROLLING_METADATA.contains(i.getAction()))
+ .limit(lookbackLimit)
+ .collect(Collectors.toList());
- log.debug("Walking back up to {} commits to find rolling metadata for keys: {}",
- lookbackLimit, remainingKeys);
+ log.debug("Walking back up to {} instants to find rolling metadata for keys: {}", lookbackLimit, remaining);
+ int instantsWalkedBack = 0;
- for (HoodieInstant instant : recentCommits) {
- if (remainingKeys.isEmpty()) {
- break; // Found all keys
+ try {
+ for (HoodieInstant instant : instants) {
+ if (remaining.isEmpty()) {
+ break;
}
+ String action = instant.getAction();
+ Map extraMeta = null;
- commitsWalkedBack++;
- HoodieCommitMetadata commitMetadata = table.getMetaClient().getActiveTimeline().readInstantContent(instant, HoodieCommitMetadata.class);
+ if (HoodieTimeline.CLEAN_ACTION.equals(action)) {
+ HoodieCleanMetadata cleanMeta = table.getActiveTimeline().readCleanMetadata(instant);
+ extraMeta = cleanMeta.getExtraMetadata();
+ } else {
+ HoodieCommitMetadata commitMeta = table.getMetaClient().getActiveTimeline()
+ .readInstantContent(instant, HoodieCommitMetadata.class);
+ extraMeta = commitMeta.getExtraMetadata();
+ }
+ instantsWalkedBack++;
- // Check for remaining keys in this commit
- for (String key : new HashSet<>(remainingKeys)) {
- String value = commitMetadata.getMetadata(key);
+ if (extraMeta == null) {
+ continue;
+ }
+ for (String key : new HashSet<>(remaining)) {
+ String value = extraMeta.get(key);
if (!StringUtils.isNullOrEmpty(value)) {
foundRollingMetadata.put(key, value);
- remainingKeys.remove(key);
- log.debug("Found rolling metadata key '{}' in commit {} with value: {}",
- key, instant.requestedTime(), value);
+ remaining.remove(key);
+ log.debug("Found rolling metadata key '{}' in {} instant {} with value: {}",
+ key, action, instant.requestedTime(), value);
}
}
}
- // Add found rolling metadata to current commit
- for (Map.Entry entry : foundRollingMetadata.entrySet()) {
- metadata.addMetadata(entry.getKey(), entry.getValue());
+ if (!foundRollingMetadata.isEmpty() || !remaining.isEmpty()) {
+ log.info("Rolling metadata: walked {} instants. Rolled forward: {}, Not found: {}, Total keys: {}",
+ instantsWalkedBack, foundRollingMetadata.size(), remaining.size(), rollingKeys.size());
+ }
+ if (!remaining.isEmpty()) {
+ log.warn("Rolling metadata keys not found in last {} instants: {}.", instantsWalkedBack, remaining);
}
+ } catch (IOException e) {
+ log.error("Failed to read previous metadata for rolling metadata keys: {}.", rollingKeys, e);
+ throw new HoodieIOException("Failed to read previous metadata for rolling keys: " + rollingKeys, e);
+ }
- int rolledForwardCount = foundRollingMetadata.size();
- int updatedCount = rollingKeys.size() - remainingKeys.size() - rolledForwardCount;
+ return foundRollingMetadata;
+ }
- if (rolledForwardCount > 0 || updatedCount > 0 || !remainingKeys.isEmpty()) {
- log.info("Rolling metadata merge completed. Walked back {} commits. "
- + "Rolled forward: {}, Updated in current: {}, Not found: {}, Total rolling keys: {}",
- commitsWalkedBack, rolledForwardCount, updatedCount, remainingKeys.size(), rollingKeys.size());
- }
+ protected Option> updateExtraMetadata(Option> extraMetadata) {
+ return CommitMetadataProperties.enrich(extraMetadata, config, context);
+ }
- if (!remainingKeys.isEmpty()) {
- log.warn("Rolling metadata keys not found in last {} commits: {}. "
- + "These keys will not be included in the current commit.", lookbackLimit, remainingKeys);
+ /**
+ * 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}.
+ *
+ * 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 stats,
+ Supplier fsViewSupplier,
+ Option> extraMetadata) {
+ if (!config.writeCommitCallbackOn()) {
+ return;
+ }
+ try {
+ if (commitCallback == null) {
+ commitCallback = HoodieCommitCallbackFactory.create(config);
}
-
- } catch (IOException e) {
- log.error("Failed to read previous commit metadata for rolling metadata keys: {}.", rollingKeys, e);
- throw new HoodieIOException("Failed to read previous commit metadata for rolling metadata keys: " + rollingKeys, e);
+ 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 17106d8d940e5..59a4f2d4db682 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
@@ -425,6 +425,8 @@ protected void completeCompaction(HoodieCommitMetadata metadata, HoodieTable tab
);
}
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);
@@ -497,6 +499,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());
}
/**
@@ -641,6 +645,8 @@ private void completeClustering(HoodieReplaceCommitMetadata replaceCommitMetadat
heartbeatClient.stop(clusteringCommitTime);
}
log.info("Clustering successfully on commit {} for table {}", clusteringCommitTime, table.getConfig().getBasePath());
+ fireCommitCallbackIfNecessary(clusteringCommitTime, HoodieTimeline.REPLACE_COMMIT_ACTION,
+ writeStats, table::getBaseFileOnlyView, Option.empty());
}
protected void runTableServicesInline(HoodieTable table, HoodieCommitMetadata metadata, Option> extraMetadata) {
@@ -726,6 +732,8 @@ Option scheduleTableServiceInternal(Option providedInstantTime,
// so it is handled differently to avoid locking for planning.
return scheduleCleaning(createTable(config, storageConf), providedInstantTime);
}
+ // Only enrich metadata after early-return checks, when we're actually going to use it
+ extraMetadata = updateExtraMetadata(extraMetadata);
Option lastCompletedInstant = lastCompletedTxnAndMetadata.isPresent()
? Option.of(lastCompletedTxnAndMetadata.get().getLeft())
: Option.empty();
@@ -1430,7 +1438,7 @@ private Option delegateToTableServiceManager(TableServiceType tableServi
case CLEAN:
return tableServiceManagerClient.executeClean();
default:
- log.info("Not supported delegate to table service manager, tableServiceType : " + tableServiceType.getAction());
+ log.info("Not supported delegate to table service manager, tableServiceType : {}", tableServiceType.getAction());
return Option.empty();
}
}
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 df06df2fbbba9..d97ea7dbb7cd1 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,10 +24,7 @@
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.HoodieWriteCommitCallback;
-import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage;
import org.apache.hudi.callback.common.WriteStatusValidator;
-import org.apache.hudi.callback.util.HoodieCommitCallbackFactory;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
import org.apache.hudi.client.heartbeat.HeartbeatUtils;
import org.apache.hudi.client.transaction.TransactionManager;
@@ -86,12 +83,14 @@
import org.apache.hudi.internal.schema.io.FileBasedInternalSchemaStorageManager;
import org.apache.hudi.internal.schema.utils.AvroSchemaEvolutionUtils;
import org.apache.hudi.internal.schema.utils.InternalSchemaUtils;
+import org.apache.hudi.internal.schema.utils.SchemaChangeUtils;
import org.apache.hudi.internal.schema.utils.SerDeHelper;
import org.apache.hudi.keygen.constant.KeyGeneratorType;
import org.apache.hudi.metadata.HoodieTableMetadataUtil;
import org.apache.hudi.metadata.HoodieTableMetadataWriter;
import org.apache.hudi.metadata.MetadataPartitionType;
import org.apache.hudi.metrics.HoodieMetrics;
+import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
@@ -145,7 +144,6 @@ public abstract class BaseHoodieWriteClient extends BaseHoodieClient
@Getter
@Setter
private transient WriteOperationType operationType;
- private transient HoodieWriteCommitCallback commitCallback;
protected transient Timer.Context writeTimer = null;
@@ -253,6 +251,7 @@ public boolean commitStats(String instantTime, TableWriteStats tableWriteStats,
if (!config.allowEmptyCommit() && tableWriteStats.isEmptyDataTableWriteStats()) {
return true;
}
+ extraMetadata = updateExtraMetadata(extraMetadata);
log.info("Committing {} action {}", instantTime, commitActionType);
// Create a Hoodie table which encapsulated the commits and files visible
HoodieTable table = hoodieTableOpt.orElse(createTable(config));
@@ -285,7 +284,7 @@ public boolean commitStats(String instantTime, TableWriteStats tableWriteStats,
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) {
@@ -301,15 +300,6 @@ public boolean commitStats(String instantTime, TableWriteStats tableWriteStats,
}
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;
}
@@ -367,7 +357,10 @@ private void saveInternalSchema(HoodieTable table, String instantTime, HoodieCom
internalSchema = InternalSchemaUtils.searchSchema(Long.parseLong(instantTime),
SerDeHelper.parseSchemas(historySchemaStr));
}
- InternalSchema evolvedSchema = AvroSchemaEvolutionUtils.reconcileSchema(schema.toAvroSchema(), internalSchema, config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS));
+ InternalSchema evolvedSchema = AvroSchemaEvolutionUtils.reconcileSchema(schema, internalSchema,
+ config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS),
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(
+ config.getStringOrDefault(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES)));
if (evolvedSchema.equals(internalSchema)) {
metadata.addMetadata(SerDeHelper.LATEST_SCHEMA, SerDeHelper.toJson(evolvedSchema));
//TODO save history schema by metaTable
@@ -637,7 +630,9 @@ public O postWrite(HoodieWriteMetadata result, String instantTime, HoodieTabl
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;
@@ -664,8 +659,37 @@ public O postWrite(HoodieWriteMetadata result, String instantTime, HoodieTabl
* @param instantTime Instant Time
* @param extraMetadata Additional Metadata passed by user
*/
- protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata, String instantTime, Option> extraMetadata) {
+ protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata, String instantTime, String commitActionType, Option> extraMetadata) {
+ try {
+ context.setJobStatus(this.getClass().getSimpleName(), "Cleaning up marker directories for commit " + instantTime + " in table "
+ + config.getTableName());
+ // Delete the marker directory for the instant.
+ 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 stats = metadata.getWriteStats();
+ fireCommitCallbackIfNecessary(instantTime, commitActionType, stats,
+ table::getBaseFileOnlyView, extraMetadata);
+ } finally {
+ this.heartbeatClient.stop(instantTime);
+ }
+ }
+
+ /**
+ * Performs post-commit cleanup when the instant is already completed and commit metadata is not
+ * available to invoke the regular post-commit hook. This can happen while recovering a streaming
+ * metadata-table write after failover. The table is recreated from the write configuration so its
+ * marker directory can still be removed, and the heartbeat is always stopped even if marker cleanup
+ * fails.
+ *
+ * @param instantTime the completed instant to clean up
+ */
+ public void postCommit(String instantTime) {
try {
+ HoodieTable table = createTable(config);
context.setJobStatus(this.getClass().getSimpleName(), "Cleaning up marker directories for commit " + instantTime + " in table "
+ config.getTableName());
// Delete the marker directory for the instant.
@@ -846,44 +870,11 @@ public void restoreToSavepoint() {
*/
public void restoreToSavepoint(String savepointTime) {
boolean initializeMetadataTableIfNecessary = config.isMetadataTableEnabled();
- if (initializeMetadataTableIfNecessary) {
- try {
- // Delete metadata table directly when users trigger savepoint rollback if mdt existed and if the savePointTime is beforeTimelineStarts
- // or before the oldest compaction on MDT.
- // We cannot restore to before the oldest compaction on MDT as we don't have the basefiles before that time.
- HoodieTableMetaClient mdtMetaClient = HoodieTableMetaClient.builder()
- .setConf(storageConf.newInstance())
- .setBasePath(getMetadataTableBasePath(config.getBasePath())).build();
- Option oldestMdtCompaction = mdtMetaClient.getCommitTimeline().filterCompletedInstants().firstInstant();
- boolean deleteMDT = false;
- if (oldestMdtCompaction.isPresent()) {
- if (LESSER_THAN_OR_EQUALS.test(savepointTime, oldestMdtCompaction.get().requestedTime())) {
- log.warn("Deleting MDT during restore to {} as the savepoint is older than oldest compaction {} on MDT",
- savepointTime, oldestMdtCompaction.get().requestedTime());
- deleteMDT = true;
- }
- }
-
- // The instant required to sync rollback to MDT has been archived and the mdt syncing will be failed
- // So that we need to delete the whole MDT here.
- if (!deleteMDT) {
- HoodieInstant syncedInstant = mdtMetaClient.createNewInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION, savepointTime);
- if (mdtMetaClient.getCommitsTimeline().isBeforeTimelineStarts(syncedInstant.requestedTime())) {
- log.warn("Deleting MDT during restore to {} as the savepoint is older than the MDT timeline {}",
- savepointTime, mdtMetaClient.getCommitsTimeline().firstInstant().get().requestedTime());
- deleteMDT = true;
- }
- }
-
- if (deleteMDT) {
- HoodieTableMetadataUtil.deleteMetadataTable(config.getBasePath(), context);
- // rollbackToSavepoint action will try to bootstrap MDT at first but sync to MDT will fail at the current scenario.
- // so that we need to disable metadata initialized here.
- initializeMetadataTableIfNecessary = false;
- }
- } catch (Exception e) {
- // Metadata directory does not exist
- }
+ if (initializeMetadataTableIfNecessary && shouldDeleteMdtBeforeRestore(savepointTime)) {
+ HoodieTableMetadataUtil.deleteMetadataTable(config.getBasePath(), context);
+ // rollbackToSavepoint action will try to bootstrap MDT at first but sync to MDT will fail at the current scenario.
+ // so that we need to disable metadata initialized here.
+ initializeMetadataTableIfNecessary = false;
}
HoodieTable table = initTable(WriteOperationType.UNKNOWN, Option.empty(), initializeMetadataTableIfNecessary);
@@ -894,6 +885,82 @@ public void restoreToSavepoint(String savepointTime) {
SavepointHelpers.validateSavepointRestore(table, savepointTime);
}
+ /**
+ * Decides whether the metadata table (MDT) must be deleted before restoring the data table to
+ * {@code targetInstant}. Returns true when restoring would leave the MDT in an inconsistent
+ * state, specifically when any of the following holds:
+ *
+ * - The target is at or before the oldest completed compaction. We cannot restore to before
+ * the oldest compaction because we don't have base files before that time.
+ * - The target is before the MDT timeline start (the relevant history was archived away).
+ *
+ * Returns false when the MDT directory does not exist or is not readable (nothing to delete or
+ * worry about). Wraps genuine IO failures ({@link IOException}) in a {@link HoodieException}
+ * so permission / network errors surface to the caller.
+ */
+ protected boolean shouldDeleteMdtBeforeRestore(String targetInstant) {
+ String mdtBasePath = getMetadataTableBasePath(config.getBasePath());
+ try {
+ // Cheap existence check first to avoid constructing an MDT meta client when there is no MDT.
+ if (!storage.exists(new StoragePath(mdtBasePath))) {
+ return false;
+ }
+ HoodieTableMetaClient mdtMetaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf.newInstance())
+ .setBasePath(mdtBasePath).build();
+ List completedCompactions = mdtMetaClient.getCommitTimeline()
+ .filterCompletedInstants().getInstants();
+ Option oldestMdtCompaction = completedCompactions.isEmpty()
+ ? Option.empty() : Option.of(completedCompactions.get(0));
+ if (oldestMdtCompaction.isPresent()
+ && LESSER_THAN_OR_EQUALS.test(targetInstant, oldestMdtCompaction.get().requestedTime())) {
+ log.warn("Deleting MDT before restore to {}: target is at or before oldest MDT compaction {}",
+ targetInstant, oldestMdtCompaction.get().requestedTime());
+ return true;
+ }
+ if (mdtMetaClient.getCommitsTimeline().isBeforeTimelineStarts(targetInstant)) {
+ log.warn("Deleting MDT before restore to {}: target is before MDT timeline start", targetInstant);
+ return true;
+ }
+ return false;
+ } catch (IOException e) {
+ throw new HoodieException(
+ "Failed to inspect MDT at " + mdtBasePath + " before restore to " + targetInstant
+ + " - refusing to silently proceed without an MDT integrity check.", e);
+ } catch (HoodieException e) {
+ // MDT directory exists but is not usable (e.g. TableNotFoundException from a partially
+ // initialized MDT). Treat as absent: no deletion needed, let the restore proceed.
+ log.warn("MDT at {} is present but could not be read ({}); skipping pre-check.",
+ mdtBasePath, e.getMessage());
+ return false;
+ }
+ }
+
+ /**
+ * Deletes the metadata table (MDT) if it would be left in an inconsistent state by a restore
+ * to {@code targetInstant}, and returns whether the MDT was actually deleted.
+ *
+ * Callers that drive restore via {@link #restoreToInstant} directly (e.g. the
+ * {@code restore_to_instant} stored procedure) should call this method before invoking
+ * {@code restoreToInstant} and suppress MDT initialization when it returns {@code true}:
+ *
+ *
{@code
+ * boolean mdtDeleted = client.deleteMdtIfNecessaryBeforeRestore(targetInstant);
+ * client.restoreToInstant(targetInstant, !mdtDeleted && enableMetadata);
+ * }
+ *
+ * @param targetInstant the instant the data table will be restored to
+ * @return {@code true} if the MDT was deleted (caller must not re-initialize it);
+ * {@code false} otherwise (MDT either did not need deletion or does not exist)
+ */
+ public boolean deleteMdtIfNecessaryBeforeRestore(String targetInstant) {
+ if (shouldDeleteMdtBeforeRestore(targetInstant)) {
+ HoodieTableMetadataUtil.deleteMetadataTable(config.getBasePath(), context);
+ return true;
+ }
+ return false;
+ }
+
@Deprecated
public boolean rollback(final String commitInstantTime) throws HoodieRollbackException {
HoodieTable table = initTable(WriteOperationType.UNKNOWN, Option.empty());
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CommitMetadataProperties.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CommitMetadataProperties.java
new file mode 100644
index 0000000000000..3a7b79812c8e7
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CommitMetadataProperties.java
@@ -0,0 +1,145 @@
+/*
+ * 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.client;
+
+import org.apache.hudi.HoodieVersion;
+import org.apache.hudi.common.config.ConfigProperty;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Enriches the {@code extraMetadata} map persisted with every commit, with version, engine, and
+ * (optionally) engine-specific properties and a configurable subset of {@link HoodieWriteConfig}
+ * values.
+ *
+ * Key namespacing:
+ *
+ * - {@code hudi.version} — writer version. Always emitted.
+ * - {@code engine} — engine type (SPARK/FLINK/JAVA). Always emitted.
+ * - Engine-supplied keys (Spark: {@code spark.*}, Java: {@code java.*}/{@code os.*}, etc.)
+ * — gated by {@link #EMBED_ENGINE_PROPERTIES_IN_COMMIT_METADATA}.
+ * - {@code config.} — values of {@link HoodieWriteConfig} entries whose keys are listed
+ * in {@link #WRITE_CONFIG_KEYS_TO_SERIALIZE_TO_COMMIT_METADATA}.
+ *
+ */
+public class CommitMetadataProperties {
+
+ static final String HUDI_VERSION_KEY = "hudi.version";
+ static final String ENGINE_KEY = "engine";
+ static final String CONFIG_KEY_PREFIX = "config.";
+
+ /**
+ * Default allowlist of write-config keys serialized into commit metadata. These are values that
+ * change across jobs/runs but aren't already captured in {@code hoodie.properties}, so they're
+ * useful for after-the-fact debugging. Intentionally excludes immutable table identity
+ * (already in {@code hoodie.properties}) and per-record/sensitive values.
+ */
+ private static final String DEFAULT_WRITE_CONFIG_KEYS = String.join(",",
+ Arrays.asList(
+ "hoodie.datasource.write.operation",
+ "hoodie.insert.shuffle.parallelism",
+ "hoodie.upsert.shuffle.parallelism",
+ "hoodie.bulkinsert.shuffle.parallelism",
+ "hoodie.delete.shuffle.parallelism",
+ "hoodie.write.concurrency.mode",
+ "hoodie.metadata.enable"));
+
+ /**
+ * When enabled, engine-specific properties supplied by
+ * {@link HoodieEngineContext#getEngineProperties()} are embedded into commit metadata for
+ * debugging (e.g. {@code spark.application.id}, {@code spark.user}). {@code hudi.version} and
+ * {@code engine} are always embedded regardless of this flag.
+ *
+ * Default is {@code false} since these add per-commit growth to the timeline. Long-running
+ * ingestion workloads writing many commits should leave this off unless debugging.
+ */
+ public static final ConfigProperty EMBED_ENGINE_PROPERTIES_IN_COMMIT_METADATA =
+ ConfigProperty
+ .key("hoodie.commit.metadata.engine.properties.embed.enable")
+ .defaultValue(false)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("When enabled, engine-specific properties (e.g. spark.application.id, "
+ + "spark.user, java.version) are embedded into commit metadata for debugging. "
+ + "hudi.version and engine name are always embedded regardless of this flag.");
+
+ /**
+ * Comma-separated list of {@link HoodieWriteConfig} keys whose values should be serialized into
+ * commit metadata under the {@code config.} prefix. Use with care: every key listed here
+ * adds an entry to every commit, which lives forever in the active and archived timeline.
+ *
+ * Empty value disables config-key serialization entirely (only {@code hudi.version} and
+ * {@code engine} are emitted).
+ */
+ public static final ConfigProperty WRITE_CONFIG_KEYS_TO_SERIALIZE_TO_COMMIT_METADATA =
+ ConfigProperty
+ .key("hoodie.write.config.keys.to.serialize.to.commit.metadata")
+ .defaultValue(DEFAULT_WRITE_CONFIG_KEYS)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("Comma-separated list of write-config keys whose values are "
+ + "serialized into the extraMetadata map of every commit (under the 'config.' "
+ + "prefix). Set to empty to skip config-key serialization entirely. Avoid adding "
+ + "keys whose values may contain credentials or large payloads, since commit "
+ + "metadata is persisted in the timeline.");
+
+ public static Option> enrich(Option> extraMetadata,
+ HoodieWriteConfig config,
+ HoodieEngineContext context) {
+ Map newMetadata = new HashMap<>();
+ if (extraMetadata.isPresent()) {
+ newMetadata.putAll(extraMetadata.get());
+ }
+
+ newMetadata.put(HUDI_VERSION_KEY, HoodieVersion.get());
+ newMetadata.put(ENGINE_KEY, config.getEngineType().name());
+
+ if (config.getBoolean(EMBED_ENGINE_PROPERTIES_IN_COMMIT_METADATA)) {
+ newMetadata.putAll(context.getEngineProperties());
+ }
+
+ for (String key : parseConfigKeys(config.getString(WRITE_CONFIG_KEYS_TO_SERIALIZE_TO_COMMIT_METADATA))) {
+ String value = config.getString(key);
+ if (!StringUtils.isNullOrEmpty(value)) {
+ newMetadata.put(CONFIG_KEY_PREFIX + key, value);
+ }
+ }
+
+ return Option.of(newMetadata);
+ }
+
+ private static List parseConfigKeys(String csv) {
+ if (StringUtils.isNullOrEmpty(csv)) {
+ return Collections.emptyList();
+ }
+ return Arrays.stream(csv.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .collect(Collectors.toList());
+ }
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
index 32be4b2741390..6abe10c6cca75 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
@@ -299,13 +299,12 @@ private List runRenamingOps(HoodieTableMetaClient metaClient,
context.setJobStatus(this.getClass().getSimpleName(), "Execute unschedule operations: " + config.getTableName());
return context.map(renameActions, lfPair -> {
try {
- log.info("RENAME " + lfPair.getLeft().getPath() + " => " + lfPair.getRight().getPath());
+ log.info("RENAME {} => {}", lfPair.getLeft().getPath(), lfPair.getRight().getPath());
renameLogFile(metaClient, lfPair.getLeft(), lfPair.getRight());
return new RenameOpResult(lfPair, true, Option.empty());
} catch (IOException e) {
log.error("Error renaming log file", e);
- log.error("\n\n\n***NOTE Compaction is in inconsistent state. Try running \"compaction repair "
- + lfPair.getLeft().getDeltaCommitTime() + "\" to recover from failure ***\n\n\n");
+ log.error("\n\n\n***NOTE Compaction is in inconsistent state. Try running \"compaction repair {}\" to recover from failure ***\n\n\n", lfPair.getLeft().getDeltaCommitTime());
return new RenameOpResult(lfPair, false, Option.of(e));
}
}, parallelism);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTableServiceManagerClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTableServiceManagerClient.java
index 4f13034c89df8..ec7dd1b30b0e8 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTableServiceManagerClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieTableServiceManagerClient.java
@@ -93,7 +93,7 @@ private String executeRequest(String requestPath, Map queryParam
queryParameters.forEach(builder::addParameter);
String url = builder.toString();
- log.info("Sending request to table management service : (" + url + ")");
+ log.info("Sending request to table management service : ({})", url);
int timeoutMs = this.config.getConnectionTimeoutSec() * 1000;
int requestRetryLimit = config.getConnectionRetryLimit();
int connectionRetryDelay = config.getConnectionRetryDelay();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapRegexModeSelector.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapRegexModeSelector.java
index 65fda8e6cf7df..6869016cf8851 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapRegexModeSelector.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapRegexModeSelector.java
@@ -48,7 +48,7 @@ public BootstrapRegexModeSelector(HoodieWriteConfig writeConfig) {
this.bootstrapModeOnMatch = writeConfig.getBootstrapModeForRegexMatch();
this.defaultMode = BootstrapMode.FULL_RECORD.equals(bootstrapModeOnMatch)
? BootstrapMode.METADATA_ONLY : BootstrapMode.FULL_RECORD;
- log.info("Default Mode :" + defaultMode + ", on Match Mode :" + bootstrapModeOnMatch);
+ log.info("Default Mode :{}, on Match Mode :{}", defaultMode, bootstrapModeOnMatch);
}
@Override
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
index 2df9d0940d5b2..7caeab7ead3d9 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
@@ -97,7 +97,7 @@ static EmbeddedTimelineService getOrStartEmbeddedTimelineService(HoodieEngineCon
synchronized (SERVICE_LOCK) {
if (RUNNING_SERVICES.containsKey(timelineServiceIdentifier)) {
RUNNING_SERVICES.get(timelineServiceIdentifier).addBasePath(writeConfig.getBasePath());
- log.info("Reusing existing embedded timeline server with configuration: " + RUNNING_SERVICES.get(timelineServiceIdentifier).serviceConfig);
+ log.info("Reusing existing embedded timeline server with configuration: {}", RUNNING_SERVICES.get(timelineServiceIdentifier).serviceConfig);
return RUNNING_SERVICES.get(timelineServiceIdentifier);
}
// if no compatible instance is found, create a new one
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java
index a043f73e632c5..b8f2f15fdf0a4 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java
@@ -19,6 +19,7 @@
package org.apache.hudi.client.heartbeat;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.CustomizedThreadFactory;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieHeartbeatException;
@@ -35,9 +36,15 @@
import java.io.OutputStream;
import java.io.Serializable;
import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import static org.apache.hudi.common.heartbeat.HoodieHeartbeatUtils.getLastHeartbeatTime;
@@ -58,7 +65,16 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable {
// heartbeat interval in millis
private final Long heartbeatIntervalInMs;
private final Long maxAllowableHeartbeatIntervalInMs;
+ // Maximum time the scheduler thread will wait for a single heartbeat file write to complete before
+ // abandoning it and letting the next tick retry. Bounded to one interval so that a slow/hung
+ // storage write cannot block the scheduler thread (and thus freeze all subsequent heartbeats).
+ private final Long heartbeatWriteTimeoutMs;
private final Map instantToHeartbeatMap;
+ // Daemon executor used to perform the (potentially slow) storage write off the scheduler thread so the
+ // write can be time-bounded. A cached pool is intentional: if one write hangs, that thread is left
+ // parked while the next tick proceeds on a fresh thread. Lazily created and marked transient since
+ // this client is Serializable with a transient storage handle.
+ private transient ExecutorService heartbeatWriteExecutor;
public HoodieHeartbeatClient(HoodieStorage storage, String basePath, Long heartbeatIntervalInMs,
Integer numTolerableHeartbeatMisses) {
@@ -68,9 +84,18 @@ public HoodieHeartbeatClient(HoodieStorage storage, String basePath, Long heartb
this.heartbeatFolderPath = HoodieTableMetaClient.getHeartbeatFolderPath(basePath);
this.heartbeatIntervalInMs = heartbeatIntervalInMs;
this.maxAllowableHeartbeatIntervalInMs = this.heartbeatIntervalInMs * numTolerableHeartbeatMisses;
+ this.heartbeatWriteTimeoutMs = this.heartbeatIntervalInMs;
this.instantToHeartbeatMap = new ConcurrentHashMap<>();
}
+ private synchronized ExecutorService getHeartbeatWriteExecutor() {
+ if (heartbeatWriteExecutor == null) {
+ heartbeatWriteExecutor =
+ Executors.newCachedThreadPool(new CustomizedThreadFactory("heartbeat_write", true));
+ }
+ return heartbeatWriteExecutor;
+ }
+
@Data
static class Heartbeat {
@@ -79,10 +104,12 @@ static class Heartbeat {
private boolean isHeartbeatStopped = false;
private Long lastHeartbeatTime;
private Integer numHeartbeats = 0;
- private Timer timer = new Timer(true);
+ private ScheduledExecutorService heartbeatScheduler =
+ Executors.newSingleThreadScheduledExecutor(new CustomizedThreadFactory("heartbeat_scheduler", true));
+ private ScheduledFuture scheduledFuture;
}
- class HeartbeatTask extends TimerTask {
+ class HeartbeatTask implements Runnable {
private final String instantTime;
@@ -92,7 +119,11 @@ class HeartbeatTask extends TimerTask {
@Override
public void run() {
- updateHeartbeat(instantTime);
+ try {
+ updateHeartbeat(instantTime);
+ } catch (Exception e) {
+ log.error("Failed to update heartbeat for instant {}; will retry on next tick", instantTime, e);
+ }
}
}
@@ -114,11 +145,11 @@ public void start(String instantTime) {
newHeartbeat.setHeartbeatStarted(true);
instantToHeartbeatMap.put(instantTime, newHeartbeat);
// Ensure heartbeat is generated for the first time with this blocking call.
- // Since timer submits the task to a thread, no guarantee when that thread will get CPU
+ // Since scheduler submits the task to a thread, no guarantee when that thread will get CPU
// cycles to generate the first heartbeat.
updateHeartbeat(instantTime);
- newHeartbeat.getTimer().scheduleAtFixedRate(new HeartbeatTask(instantTime), this.heartbeatIntervalInMs,
- this.heartbeatIntervalInMs);
+ newHeartbeat.setScheduledFuture(newHeartbeat.getHeartbeatScheduler().scheduleAtFixedRate(
+ new HeartbeatTask(instantTime), this.heartbeatIntervalInMs, this.heartbeatIntervalInMs, TimeUnit.MILLISECONDS));
}
/**
@@ -130,7 +161,7 @@ public void start(String instantTime) {
public Heartbeat stop(String instantTime) throws HoodieException {
Heartbeat heartbeat = instantToHeartbeatMap.remove(instantTime);
if (isHeartbeatStarted(heartbeat)) {
- stopHeartbeatTimer(heartbeat);
+ stopHeartbeatScheduler(heartbeat);
HeartbeatUtils.deleteHeartbeatFile(storage, basePath, instantTime);
log.info("Deleted heartbeat file for instant {}", instantTime);
}
@@ -138,12 +169,12 @@ public Heartbeat stop(String instantTime) throws HoodieException {
}
/**
- * Stops all timers of heartbeats started via this instance of the client.
+ * Stops all heartbeat schedulers started via this instance of the client.
*
* @throws HoodieException
*/
public void stopHeartbeatTimers() throws HoodieException {
- instantToHeartbeatMap.values().stream().filter(this::isHeartbeatStarted).forEach(this::stopHeartbeatTimer);
+ instantToHeartbeatMap.values().stream().filter(this::isHeartbeatStarted).forEach(this::stopHeartbeatScheduler);
}
/**
@@ -158,17 +189,24 @@ private boolean isHeartbeatStarted(Heartbeat heartbeat) {
}
/**
- * Stops the timer of the given heartbeat.
+ * Stops the scheduler of the given heartbeat.
*
* @param heartbeat The heartbeat to stop.
*/
- private void stopHeartbeatTimer(Heartbeat heartbeat) {
+ private void stopHeartbeatScheduler(Heartbeat heartbeat) {
log.info("Stopping heartbeat for instant {}", heartbeat.getInstantTime());
- heartbeat.getTimer().cancel();
+ shutdownHeartbeatScheduler(heartbeat);
heartbeat.setHeartbeatStopped(true);
log.info("Stopped heartbeat for instant {}", heartbeat.getInstantTime());
}
+ private void shutdownHeartbeatScheduler(Heartbeat heartbeat) {
+ if (heartbeat.getScheduledFuture() != null) {
+ heartbeat.getScheduledFuture().cancel(false);
+ }
+ heartbeat.getHeartbeatScheduler().shutdownNow();
+ }
+
public static Boolean heartbeatExists(HoodieStorage storage, String basePath, String instantTime) throws IOException {
StoragePath heartbeatFilePath = new StoragePath(
HoodieTableMetaClient.getHeartbeatFolderPath(basePath), instantTime);
@@ -178,17 +216,18 @@ public static Boolean heartbeatExists(HoodieStorage storage, String basePath, St
public boolean isHeartbeatExpired(String instantTime) throws IOException {
Long currentTime = System.currentTimeMillis();
Heartbeat lastHeartbeatForWriter = instantToHeartbeatMap.get(instantTime);
- if (lastHeartbeatForWriter == null) {
- log.info("Heartbeat not found in internal map, falling back to reading from DFS");
- long lastHeartbeatForWriterTime = getLastHeartbeatTime(this.storage, basePath, instantTime);
- lastHeartbeatForWriter = new Heartbeat();
- lastHeartbeatForWriter.setLastHeartbeatTime(lastHeartbeatForWriterTime);
- lastHeartbeatForWriter.setInstantTime(instantTime);
- lastHeartbeatForWriter.getTimer().cancel();
+ Long lastHeartbeatTime = lastHeartbeatForWriter == null ? null : lastHeartbeatForWriter.getLastHeartbeatTime();
+ // lastHeartbeatTime can be null when the heartbeat is not in the internal map, or when it is in the
+ // map but no heartbeat has been generated yet (e.g. the first write timed out). In both cases fall
+ // back to reading the last heartbeat time from DFS (returns 0 if no heartbeat file exists, which is
+ // correctly treated as expired).
+ if (lastHeartbeatTime == null) {
+ log.info("Heartbeat time not available in internal map, falling back to reading from DFS");
+ lastHeartbeatTime = getLastHeartbeatTime(this.storage, basePath, instantTime);
}
- if (currentTime - lastHeartbeatForWriter.getLastHeartbeatTime() > this.maxAllowableHeartbeatIntervalInMs) {
+ if (currentTime - lastHeartbeatTime > this.maxAllowableHeartbeatIntervalInMs) {
log.warn("Heartbeat expired, currentTime = {}, last heartbeat = {}, heartbeat interval = {}", currentTime,
- lastHeartbeatForWriter, this.heartbeatIntervalInMs);
+ lastHeartbeatTime, this.heartbeatIntervalInMs);
return true;
}
return false;
@@ -197,20 +236,31 @@ public boolean isHeartbeatExpired(String instantTime) throws IOException {
private void updateHeartbeat(String instantTime) throws HoodieHeartbeatException {
try {
Long newHeartbeatTime = System.currentTimeMillis();
- OutputStream outputStream =
- this.storage.create(
- new StoragePath(heartbeatFolderPath, instantTime), true);
- outputStream.close();
+ writeHeartbeatFile(instantTime);
Heartbeat heartbeat = instantToHeartbeatMap.get(instantTime);
if (heartbeat.getLastHeartbeatTime() != null && isHeartbeatExpired(instantTime)) {
- log.error("Aborting, missed generating heartbeat within allowable interval {} ms", this.maxAllowableHeartbeatIntervalInMs);
- // Since TimerTask allows only java.lang.Runnable, cannot throw an exception and bubble to the caller thread, hence
- // explicitly interrupting the timer thread.
- Thread.currentThread().interrupt();
+ // A previous refresh was delayed past the tolerable interval. Stop refreshing this heartbeat
+ // (cancel the scheduler) and do NOT advance the last heartbeat time, so the heartbeat stays expired
+ // and the writer aborts at commit time via HeartbeatUtils.abortIfHeartbeatExpired(). We must not
+ // keep refreshing here: a concurrent process (e.g. an async cleaner under LAZY failed-writes
+ // policy) may already have started rolling back this instant once it observed the expiry, and
+ // resurrecting the heartbeat could let this writer commit on top of rolled-back files.
+ // The scheduler is cancelled cleanly rather than via Thread.interrupt(), which would permanently
+ // kill the scheduler thread (turning a transient delay into a permanent blackout on the first miss).
+ log.error("Missed generating heartbeat for instant {} within allowable interval {} ms; stopping heartbeat refresh",
+ instantTime, this.maxAllowableHeartbeatIntervalInMs);
+ shutdownHeartbeatScheduler(heartbeat);
+ return;
}
heartbeat.setInstantTime(instantTime);
heartbeat.setLastHeartbeatTime(newHeartbeatTime);
heartbeat.setNumHeartbeats(heartbeat.getNumHeartbeats() + 1);
+ } catch (TimeoutException te) {
+ // The storage write did not complete within the bounded window. Do not advance the last heartbeat
+ // time (the write is unconfirmed); the next scheduled tick will retry on a fresh executor thread.
+ // Crucially, the scheduler thread is freed instead of being blocked by a hung storage call.
+ log.warn("Heartbeat file write for instant {} did not complete within {} ms; will retry on next tick",
+ instantTime, this.heartbeatWriteTimeoutMs);
} catch (IOException io) {
boolean isHeartbeatStopped = instantToHeartbeatMap.get(instantTime).isHeartbeatStopped();
if (isHeartbeatStopped) {
@@ -221,13 +271,49 @@ private void updateHeartbeat(String instantTime) throws HoodieHeartbeatException
}
}
+ /**
+ * Writes the heartbeat file for the given instant on a dedicated daemon executor, bounded by
+ * {@link #heartbeatWriteTimeoutMs}. Performing the storage write off the scheduler thread (and with a
+ * timeout) ensures that a slow or hung storage call cannot block the scheduler thread and freeze all
+ * subsequent heartbeats for this instant.
+ */
+ private void writeHeartbeatFile(String instantTime) throws IOException, TimeoutException {
+ Future future = getHeartbeatWriteExecutor().submit(() -> {
+ try (OutputStream outputStream =
+ this.storage.create(new StoragePath(heartbeatFolderPath, instantTime), true)) {
+ // create + close confirms the heartbeat file write landed on storage.
+ }
+ return null;
+ });
+ try {
+ future.get(heartbeatWriteTimeoutMs, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException te) {
+ future.cancel(true);
+ throw te;
+ } catch (InterruptedException ie) {
+ future.cancel(true);
+ Thread.currentThread().interrupt();
+ throw new HoodieHeartbeatException("Interrupted while writing heartbeat for instant " + instantTime, ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof IOException) {
+ throw (IOException) cause;
+ }
+ throw new HoodieHeartbeatException("Failed to write heartbeat for instant " + instantTime, cause);
+ }
+ }
+
public Heartbeat getHeartbeat(String instantTime) {
return this.instantToHeartbeatMap.get(instantTime);
}
@Override
- public void close() {
+ public synchronized void close() {
this.stopHeartbeatTimers();
this.instantToHeartbeatMap.clear();
+ if (heartbeatWriteExecutor != null) {
+ heartbeatWriteExecutor.shutdownNow();
+ heartbeatWriteExecutor = null;
+ }
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v1/TimelineArchiverV1.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v1/TimelineArchiverV1.java
index d518ac5525dd6..579b2a8c6c87d 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v1/TimelineArchiverV1.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v1/TimelineArchiverV1.java
@@ -31,8 +31,8 @@
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.HoodieLogFormat.Writer;
+import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
@@ -116,9 +116,12 @@ public TimelineArchiverV1(HoodieWriteConfig config, HoodieTable tabl
private Writer openWriter(StoragePath archivePath) {
try {
if (this.writer == null) {
- return HoodieLogFormat.newWriterBuilder().onParentPath(archivePath).withInstantTime("")
- .withFileId(archiveFilePath.getName()).withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION)
- .withStorage(metaClient.getStorage()).build();
+ return HoodieLogFormatWriter.builder()
+ .withParentPath(archivePath).withInstantTime("")
+ .withLogFileId(archiveFilePath.getName())
+ .withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION)
+ .withStorage(metaClient.getStorage())
+ .build();
} else {
return this.writer;
}
@@ -356,7 +359,7 @@ private List getInstantsToArchive() throws IOException {
log.info("Not archiving as there is no compaction yet on the metadata table");
instants = Stream.empty();
} else {
- log.info("Limiting archiving of instants to latest compaction on metadata table at " + latestCompactionTime.get());
+ log.info("Limiting archiving of instants to latest compaction on metadata table at {}", latestCompactionTime.get());
instants = instants.filter(instant -> compareTimestamps(instant.requestedTime(), LESSER_THAN,
latestCompactionTime.get()));
}
@@ -416,7 +419,7 @@ private List getInstantsToArchive() throws IOException {
}
private boolean deleteArchivedInstants(List archivedInstants, HoodieEngineContext context) throws IOException {
- log.info("Deleting instants " + archivedInstants);
+ log.info("Deleting instants {}", archivedInstants);
List pendingInstants = new ArrayList<>();
List completedInstants = new ArrayList<>();
@@ -460,7 +463,7 @@ private boolean deleteArchivedInstants(List archivedInstants, Hoo
public void archive(HoodieEngineContext context, List instants) throws HoodieCommitException {
try {
Schema wrapperSchema = HoodieArchivedMetaEntry.getClassSchema();
- log.info("Wrapper schema " + wrapperSchema.toString());
+ log.info("Wrapper schema {}", wrapperSchema);
List records = new ArrayList<>();
for (HoodieInstant hoodieInstant : instants) {
try {
@@ -471,7 +474,7 @@ public void archive(HoodieEngineContext context, List instants) t
}
} catch (Exception e) {
InstantFileNameGenerator fileNameFactory = new InstantFileNameGeneratorV1();
- log.error("Failed to archive commits, .commit file: " + fileNameFactory.getFileName(hoodieInstant), e);
+ log.error("Failed to archive commits, .commit file: {}", fileNameFactory.getFileName(hoodieInstant), e);
if (this.config.isFailOnTimelineArchivingEnabled()) {
throw e;
}
@@ -486,7 +489,7 @@ public void archive(HoodieEngineContext context, List instants) t
private void deleteAnyLeftOverMarkers(HoodieEngineContext context, HoodieInstant instant) {
WriteMarkers writeMarkers = WriteMarkersFactory.get(config.getMarkersType(), table, instant.requestedTime());
if (writeMarkers.deleteMarkerDir(context, config.getMarkersDeleteParallelism())) {
- log.info("Cleaned up left over marker directory for instant :" + instant);
+ log.info("Cleaned up left over marker directory for instant :{}", instant);
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v2/LSMTimelineWriter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v2/LSMTimelineWriter.java
index b555a0646183c..97b2dc01d3acc 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v2/LSMTimelineWriter.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v2/LSMTimelineWriter.java
@@ -49,7 +49,6 @@
import org.apache.hudi.table.HoodieTable;
import lombok.extern.slf4j.Slf4j;
-import org.apache.avro.Schema;
import org.apache.avro.generic.IndexedRecord;
import java.io.IOException;
@@ -137,9 +136,8 @@ public void write(
throw new HoodieIOException("Failed to check archiving file before write: " + filePath, ioe);
}
try (HoodieFileWriter writer = openWriter(filePath)) {
- Schema wrapperSchema = HoodieLSMTimelineInstant.getClassSchema();
- log.info("Writing schema " + wrapperSchema.toString());
- HoodieSchema schema = HoodieSchema.fromAvroSchema(wrapperSchema);
+ HoodieSchema schema = HoodieSchema.fromAvroSchema(HoodieLSMTimelineInstant.getClassSchema());
+ log.info("Writing schema {}", schema);
for (ActiveAction activeAction : activeActions) {
try {
preWriteCallback.ifPresent(callback -> callback.accept(activeAction));
@@ -147,7 +145,7 @@ public void write(
final HoodieLSMTimelineInstant metaEntry = MetadataConversionUtils.createLSMTimelineInstant(activeAction, metaClient);
writer.write(metaEntry.getInstantTime(), new HoodieAvroIndexedRecord(metaEntry), schema);
} catch (Exception e) {
- log.error("Failed to write instant: " + activeAction.getInstantTime(), e);
+ log.error("Failed to write instant: {}", activeAction.getInstantTime(), e);
exceptionHandler.ifPresent(handler -> handler.accept(e));
}
}
@@ -290,7 +288,7 @@ private Option doCompact(HoodieLSMTimelineManifest manifest, int layer)
compactFiles(candidateFiles, compactedFileName);
// 4. update the manifest file
updateManifest(candidateFiles, compactedFileName);
- log.info("Finishes compaction of source files: " + candidateFiles);
+ log.info("Finishes compaction of source files: {}", candidateFiles);
return Option.of(compactedFileName);
}
return Option.empty();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/BucketIndexConcurrentFileWritesConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/BucketIndexConcurrentFileWritesConflictResolutionStrategy.java
index 54112abd75eb2..01b071f714a17 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/BucketIndexConcurrentFileWritesConflictResolutionStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/BucketIndexConcurrentFileWritesConflictResolutionStrategy.java
@@ -51,8 +51,7 @@ public boolean hasConflict(ConcurrentOperation thisOperation, ConcurrentOperatio
Set intersection = new HashSet<>(partitionBucketIdSetForFirstInstant);
intersection.retainAll(partitionBucketIdSetForSecondInstant);
if (!intersection.isEmpty()) {
- log.info("Found conflicting writes between first operation = " + thisOperation
- + ", second operation = " + otherOperation + " , intersecting bucket ids " + intersection);
+ log.info("Found conflicting writes between first operation = {}, second operation = {} , intersecting bucket ids {}", thisOperation, otherOperation, intersection);
return true;
}
return false;
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentSchemaEvolutionTableSchemaGetter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentSchemaEvolutionTableSchemaGetter.java
index 0bb7db3fa5833..3ff8625e7457f 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentSchemaEvolutionTableSchemaGetter.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentSchemaEvolutionTableSchemaGetter.java
@@ -25,7 +25,7 @@
import org.apache.hudi.common.table.TableSchemaResolver;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.TimelineLayout;
+import org.apache.hudi.common.table.timeline.InstantComparator;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
@@ -60,6 +60,8 @@ class ConcurrentSchemaEvolutionTableSchemaGetter {
private final Lazy> tableSchemaCache;
+ private final InstantComparator instantComparator;
+
private Option latestCommitWithValidSchema = Option.empty();
@VisibleForTesting
@@ -69,10 +71,18 @@ public ConcurrentHashMap getTableSchemaCache() {
public ConcurrentSchemaEvolutionTableSchemaGetter(HoodieTableMetaClient metaClient) {
this.metaClient = metaClient;
+ this.instantComparator = metaClient.getTimelineLayout().getInstantComparator();
// Unbounded sized map. Should replace with some caching library.
this.tableSchemaCache = Lazy.lazily(ConcurrentHashMap::new);
}
+ /**
+ * Returns the timestamp ordering the instant in the schema evolution timeline.
+ */
+ String getOrderingTime(HoodieInstant instant) {
+ return instantComparator.getOrderingTime(instant);
+ }
+
/**
* Handles partition column logic for a given schema.
*
@@ -160,9 +170,11 @@ Option> getLastCommitMetadataWithValidSchemaFr
// the timeline finding a completed instant containing a valid schema.
ConcurrentHashMap tableSchemaAtInstant = new ConcurrentHashMap<>();
Option instantWithTableSchema = Option.fromJavaOptional(reversedTimelineStream
- // If a completion time is specified, find the first eligible instant in the schema evolution timeline.
- // Should switch to completion time based.
- .filter(s -> instant.isEmpty() || compareTimestamps(s.getCompletionTime(), LESSER_THAN_OR_EQUALS, instant.get().getCompletionTime()))
+ // Find the first eligible instant whose ordering time is no later than the target instant's;
+ // a target instant without an ordering time (not completed yet, on table version 8 and above)
+ // does not bound the lookup.
+ .filter(s -> instant.isEmpty() || StringUtils.isNullOrEmpty(getOrderingTime(instant.get()))
+ || compareTimestamps(getOrderingTime(s), LESSER_THAN_OR_EQUALS, getOrderingTime(instant.get())))
// Make sure the commit metadata has a valid schema inside. Same caching the result for expensive operation.
.filter(s -> {
try {
@@ -193,6 +205,8 @@ Option> getLastCommitMetadataWithValidSchemaFr
/**
* Get timeline in REVERSE order that only contains completed instants which POTENTIALLY evolve the table schema.
+ * The stream follows the timeline layout's instant ordering, newest first (completion time for
+ * layout v2, requested time for v1).
* For types of instants that are included and not reflecting table schema at their instant completion time please refer
* comments inside the code.
*/
@@ -214,9 +228,7 @@ public Stream computeSchemaEvolutionTimelineInReverseOrder() {
}
// We only care committed instant when it comes to table schema.
- TimelineLayout timelineLayout = metaClient.getTimelineLayout();
- // Table schema getter is completion time based ordering.
- Comparator reversedComparator = timelineLayout.getInstantComparator().completionTimeOrderedComparator().reversed();
+ Comparator reversedComparator = instantComparator.orderingComparator().reversed();
// The timeline still contains DELTA_COMMIT_ACTION/COMMIT_ACTION which might not contain a valid schema
// field in their commit metadata.
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/DirectMarkerTransactionManager.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/DirectMarkerTransactionManager.java
index 02b027f12d31f..90c5e963da1e2 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/DirectMarkerTransactionManager.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/DirectMarkerTransactionManager.java
@@ -48,22 +48,20 @@ public DirectMarkerTransactionManager(HoodieWriteConfig config, HoodieStorage st
public void beginTransaction(String newTxnOwnerInstantTime, InstantGenerator instantGenerator) {
if (isLockRequired) {
- LOG.info("Transaction starting for " + newTxnOwnerInstantTime + " and " + filePath);
+ LOG.info("Transaction starting for {} and {}", newTxnOwnerInstantTime, filePath);
lockManager.lock();
reset(changeActionInstant, Option.of(getInstant(newTxnOwnerInstantTime, instantGenerator)), Option.empty());
- LOG.info("Transaction started for " + newTxnOwnerInstantTime + " and " + filePath);
+ LOG.info("Transaction started for {} and {}", newTxnOwnerInstantTime, filePath);
}
}
public void endTransaction(String currentTxnOwnerInstantTime, InstantGenerator instantGenerator) {
if (isLockRequired) {
- LOG.info("Transaction ending with transaction owner " + currentTxnOwnerInstantTime
- + " for " + filePath);
+ LOG.info("Transaction ending with transaction owner {} for {}", currentTxnOwnerInstantTime, filePath);
if (reset(Option.of(getInstant(currentTxnOwnerInstantTime, instantGenerator)), Option.empty(), Option.empty())) {
lockManager.unlock();
- LOG.info("Transaction ended with transaction owner " + currentTxnOwnerInstantTime
- + " for " + filePath);
+ LOG.info("Transaction ended with transaction owner {} for {}", currentTxnOwnerInstantTime, filePath);
}
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
index e2eaa53103036..92c6f6f66ba54 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
@@ -142,8 +142,7 @@ public boolean hasConflict(ConcurrentOperation thisOperation, ConcurrentOperatio
Set> intersection = new HashSet<>(partitionAndFileIdsSetForFirstInstant);
intersection.retainAll(partitionAndFileIdsSetForSecondInstant);
if (!intersection.isEmpty()) {
- log.info("Found conflicting writes between first operation = " + thisOperation
- + ", second operation = " + otherOperation + " , intersecting file ids " + intersection);
+ log.info("Found conflicting writes between first operation = {}, second operation = {} , intersecting file ids {}", thisOperation, otherOperation, intersection);
return true;
}
return false;
@@ -163,8 +162,7 @@ private boolean isRollbackConflict(ConcurrentOperation thisOperation, Concurrent
String rolledbackCommit = otherOperation.getRolledbackCommit();
String thisCommitTimestamp = thisOperation.getInstantTimestamp();
if (rolledbackCommit != null && rolledbackCommit.equals(thisCommitTimestamp)) {
- log.error("Found rollback conflict: rollback operation " + otherOperation
- + " is rolling back commit " + thisCommitTimestamp + " created by operation " + thisOperation);
+ log.error("Found rollback conflict: rollback operation {} is rolling back commit {} created by operation {}", otherOperation, thisCommitTimestamp, thisOperation);
return true;
}
}
@@ -202,8 +200,66 @@ public Option resolveConflict(HoodieTable table,
return thisOperation.getCommitMetadataOption();
}
// just abort the current write if conflicts are found (failed for rollback conflicts).
- throw new HoodieWriteConflictException(new ConcurrentModificationException("Cannot resolve conflicts for overlapping writes between first operation = " + thisOperation
- + ", second operation = " + otherOperation));
+ throw new HoodieWriteConflictException(new ConcurrentModificationException(buildConflictErrorMessage(thisOperation, otherOperation)));
+ }
+
+ /**
+ * Builds a detailed error message for write conflicts based on the operation types involved.
+ */
+ private String buildConflictErrorMessage(ConcurrentOperation thisOperation, ConcurrentOperation otherOperation) {
+ boolean thisIsTableService = WriteOperationType.isTableService(thisOperation.getOperationType());
+ boolean otherIsTableService = WriteOperationType.isTableService(otherOperation.getOperationType());
+ String thisOperationDescription = formatOperationDescription(thisOperation);
+ String otherOperationDescription = formatOperationDescription(otherOperation);
+ // If either operation is a table service, provide specific retry guidance
+ if (thisIsTableService || otherIsTableService) {
+ ConcurrentOperation tableServiceOperation = thisIsTableService ? thisOperation : otherOperation;
+ String tableServiceDescription = thisIsTableService ? thisOperationDescription : otherOperationDescription;
+ String regularOperationDescription = thisIsTableService ? otherOperationDescription : thisOperationDescription;
+ String serviceType = getTableServiceDisplayName(tableServiceOperation.getOperationType());
+ return String.format(
+ "Cannot resolve conflicts for overlapping writes. %s is currently running and has overlapping file groups with %s. "
+ + "Please retry the write operation after the %s completes.",
+ tableServiceDescription, regularOperationDescription, serviceType.toLowerCase()
+ );
+ }
+ // For regular write operations conflicting with each other
+ return String.format(
+ "Cannot resolve conflicts for overlapping writes. %s has overlapping file groups with %s.",
+ thisOperationDescription, otherOperationDescription
+ );
+ }
+
+ /**
+ * Formats a description of an operation including its type, instant, and state.
+ */
+ private String formatOperationDescription(ConcurrentOperation operation) {
+ String operationName = WriteOperationType.isTableService(operation.getOperationType())
+ ? "Table " + getTableServiceDisplayName(operation.getOperationType())
+ : operation.getOperationType().value() + " operation";
+
+ return String.format("%s (instant: %s, state: %s)",
+ operationName,
+ operation.getInstantTimestamp(),
+ operation.getInstantActionState());
+ }
+
+ /**
+ * Returns a user-friendly display name for table service operations.
+ */
+ private String getTableServiceDisplayName(WriteOperationType operationType) {
+ switch (operationType) {
+ case COMPACT:
+ return "Compaction";
+ case CLUSTER:
+ return "Clustering";
+ case LOG_COMPACT:
+ return "Log Compaction";
+ case INDEX:
+ return "Indexing";
+ default:
+ return operationType.value();
+ }
}
@Override
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleSchemaConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleSchemaConflictResolutionStrategy.java
index cfcd26362552c..523b21356094c 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleSchemaConflictResolutionStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleSchemaConflictResolutionStrategy.java
@@ -30,8 +30,6 @@
import lombok.extern.slf4j.Slf4j;
-import java.util.stream.Stream;
-
import static org.apache.hudi.client.transaction.SchemaConflictResolutionStrategy.throwConcurrentSchemaEvolutionException;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
import static org.apache.hudi.common.table.timeline.InstantComparison.LESSER_THAN_OR_EQUALS;
@@ -77,7 +75,7 @@ public Option resolveConcurrentSchemaEvolution(
// schema and writer schema.
HoodieInstant lastCompletedInstantAtTxnStart = lastCompletedTxnOwnerInstant.isPresent()
? getInstantInTimelineImmediatelyPriorToTimestamp(
- lastCompletedTxnOwnerInstant.get().getCompletionTime(), schemaResolver.computeSchemaEvolutionTimelineInReverseOrder()).orElse(null)
+ schemaResolver.getOrderingTime(lastCompletedTxnOwnerInstant.get()), schemaResolver).orElse(null)
: null;
// If lastCompletedInstantAtTxnValidation is null there are 2 possibilities:
// - No committed txn at validation starts
@@ -157,9 +155,9 @@ public Option resolveConcurrentSchemaEvolution(
}
private Option getInstantInTimelineImmediatelyPriorToTimestamp(
- String timestamp, Stream reverseOrderTimeline) {
- return Option.fromJavaOptional(reverseOrderTimeline
- .filter(s -> compareTimestamps(s.getCompletionTime(), LESSER_THAN_OR_EQUALS, timestamp))
+ String timestamp, ConcurrentSchemaEvolutionTableSchemaGetter schemaResolver) {
+ return Option.fromJavaOptional(schemaResolver.computeSchemaEvolutionTimelineInReverseOrder()
+ .filter(s -> compareTimestamps(schemaResolver.getOrderingTime(s), LESSER_THAN_OR_EQUALS, timestamp))
.findFirst());
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/BaseZookeeperBasedLockProvider.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/BaseZookeeperBasedLockProvider.java
index d5b04c15c005e..6cdee60e2d1b1 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/BaseZookeeperBasedLockProvider.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/BaseZookeeperBasedLockProvider.java
@@ -68,6 +68,7 @@ public BaseZookeeperBasedLockProvider(final LockConfiguration lockConfiguration,
this.lockConfiguration = lockConfiguration;
zkBasePath = getZkBasePath(lockConfiguration);
lockKey = getLockKey(lockConfiguration);
+ int connectionTimeoutMs = ConfigUtils.getIntWithAltKeys(lockConfiguration.getConfig(), ZK_CONNECTION_TIMEOUT_MS);
this.curatorFrameworkClient = CuratorFrameworkFactory.builder()
.connectString(ConfigUtils.getStringWithAltKeys(lockConfiguration.getConfig(), ZK_CONNECT_URL))
.retryPolicy(new BoundedExponentialBackoffRetry(
@@ -75,10 +76,32 @@ public BaseZookeeperBasedLockProvider(final LockConfiguration lockConfiguration,
ConfigUtils.getIntWithAltKeys(lockConfiguration.getConfig(), LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS),
ConfigUtils.getIntWithAltKeys(lockConfiguration.getConfig(), LOCK_ACQUIRE_NUM_RETRIES)))
.sessionTimeoutMs(ConfigUtils.getIntWithAltKeys(lockConfiguration.getConfig(), ZK_SESSION_TIMEOUT_MS))
- .connectionTimeoutMs(ConfigUtils.getIntWithAltKeys(lockConfiguration.getConfig(), ZK_CONNECTION_TIMEOUT_MS))
+ .connectionTimeoutMs(connectionTimeoutMs)
.build();
this.curatorFrameworkClient.start();
- createPathIfNotExists();
+ // Once started, the Curator client owns background threads. If anything below throws, the
+ // constructor never returns the instance, so the caller can never invoke close() - clean up here.
+ try {
+ if (!this.curatorFrameworkClient.blockUntilConnected(connectionTimeoutMs, TimeUnit.MILLISECONDS)) {
+ throw new HoodieLockException("Failed to connect to ZooKeeper within " + connectionTimeoutMs + " ms");
+ }
+ createPathIfNotExists();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ closeQuietly();
+ throw new HoodieLockException("Interrupted while waiting to connect to ZooKeeper", e);
+ } catch (RuntimeException e) {
+ closeQuietly();
+ throw e;
+ }
+ }
+
+ private void closeQuietly() {
+ try {
+ this.curatorFrameworkClient.close();
+ } catch (Exception ex) {
+ log.warn("Failed to close ZooKeeper client after failed initialization", ex);
+ }
}
protected abstract String getZkBasePath(LockConfiguration lockConfiguration);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java
index fa7fde5175083..7c9362d34e404 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java
@@ -169,17 +169,15 @@ private boolean checkIfExpired() {
return true;
}
} catch (IOException | HoodieIOException e) {
- log.error(generateLogStatement(LockState.ALREADY_RELEASED) + " failed to get lockFile's modification time", e);
+ log.error("{} failed to get lockFile's modification time", generateLogStatement(LockState.ALREADY_RELEASED), e);
}
return false;
}
private void acquireLock() {
try (OutputStream os = storage.create(this.lockFile, false)) {
- if (!storage.exists(this.lockFile)) {
- initLockInfo();
- os.write(StringUtils.getUTF8Bytes(lockInfo.toString()));
- }
+ initLockInfo();
+ os.write(StringUtils.getUTF8Bytes(lockInfo.toString()));
} catch (IOException e) {
throw new HoodieIOException(generateLogStatement(LockState.FAILED_TO_ACQUIRE), e);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/LockManager.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/LockManager.java
index 21eb5da615758..6be1ebd7c911a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/LockManager.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/LockManager.java
@@ -109,7 +109,7 @@ public void unlock() {
public synchronized LockProvider getLockProvider() {
// Perform lazy initialization of lock provider only if needed
if (lockProvider == null) {
- log.info("LockProvider " + writeConfig.getLockProviderClass());
+ log.info("LockProvider {}", writeConfig.getLockProviderClass());
// Try to load lock provider with HoodieLockMetrics constructor first
Class[] metricsConstructorTypes = {LockConfiguration.class, StorageConfiguration.class, HoodieLockMetrics.class};
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java
index 2ba31e36897da..a2c6aec8afa79 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java
@@ -637,16 +637,22 @@ protected synchronized boolean renewLock() {
hoodieLockMetrics.ifPresent(HoodieLockMetrics::updateLockThrottledMetric);
// Let heartbeat retry later.
return true;
- case SUCCESS:
- // Only positive outcome
- this.setLock(currentLock.getRight().get());
- hoodieLockMetrics.ifPresent(metrics -> metrics.updateLockExpirationDeadlineMetric(
- (int) (oldExpirationMs - getCurrentEpochMs())));
- logger.info("Owner {}: Lock renewal successful. The renewal completes {} ms before expiration for lock {}.",
- ownerId, oldExpirationMs - getCurrentEpochMs(), lockFilePath);
+ case SUCCESS: {
+ // Only positive outcome. Source the deadline metric and log from the renewed lock file
+ // returned by the storage client (same as the acquisition path), not the locally
+ // computed expiration, so both callers agree on where the deadline comes from.
+ StorageLockFile renewedLock = currentLock.getRight().get();
+ this.setLock(renewedLock);
+ // Read the clock once so the metric and the log line below report the same deadline.
+ long renewalCompletionMs = getCurrentEpochMs();
+ long remainingLeaseMs = renewedLock.getValidUntilMs() - renewalCompletionMs;
+ hoodieLockMetrics.ifPresent(metrics -> metrics.updateLockExpirationDeadlineMetric((int) remainingLeaseMs));
+ logger.info("Owner {}: Lock renewal successful. The renewal completes {} ms before old expiration. The lock will expire in {} ms for lock {}.",
+ ownerId, oldExpirationMs - renewalCompletionMs, remainingLeaseMs, lockFilePath);
recordAuditOperation(AuditOperationState.RENEW, acquisitionTimestamp);
// Let heartbeat continue to renew lock lease again later.
return true;
+ }
default:
throw new HoodieLockException("Unexpected lock update result: " + currentLock.getLeft());
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LazyIterableIterator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LazyIterableIterator.java
index b921c6ddfc813..64a92ee1ae8ca 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LazyIterableIterator.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LazyIterableIterator.java
@@ -28,7 +28,7 @@
* Provide a way to obtain a inputItr of type O (output), out of an inputItr of type I (input)
*
* Things to remember: - Assumes Spark calls hasNext() to check for elements, before calling next() to obtain them -
- * Assumes hasNext() gets called atleast once. - Concrete Implementation is responsible for calling inputIterator.next()
+ * Assumes hasNext() gets called at least once. - Concrete Implementation is responsible for calling inputIterator.next()
* and doing the processing in computeNext()
*/
public abstract class LazyIterableIterator implements Iterable, Iterator {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
index 6b5ac8c575aa4..cca0486799fdd 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
@@ -89,8 +89,7 @@ public static Option resolveWriteConflictIfAny(
try {
ConcurrentOperation otherOperation = new ConcurrentOperation(instant, table.getMetaClient());
if (resolutionStrategy.hasConflict(thisOperation, otherOperation)) {
- log.info("Conflict encountered between current instant = " + thisOperation + " and instant = "
- + otherOperation + ", attempting to resolve it...");
+ log.info("Conflict encountered between current instant = {} and instant = {}, attempting to resolve it...", thisOperation, otherOperation);
resolutionStrategy.resolveConflict(table, thisOperation, otherOperation);
}
} catch (IOException io) {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/validator/StreamingOffsetValidator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/validator/StreamingOffsetValidator.java
index ce577d84ca018..0313d57c30c71 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/validator/StreamingOffsetValidator.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/validator/StreamingOffsetValidator.java
@@ -20,11 +20,13 @@
package org.apache.hudi.client.validator;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.util.CheckpointUtils;
import org.apache.hudi.common.util.CheckpointUtils.CheckpointFormat;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodiePreCommitValidatorConfig;
import org.apache.hudi.config.HoodiePreCommitValidatorConfig.ValidationFailurePolicy;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieValidationException;
import lombok.extern.slf4j.Slf4j;
@@ -50,7 +52,11 @@
*
* Subclasses specify:
* - Checkpoint format (SPARK_KAFKA, FLINK_KAFKA, etc.)
- * - Checkpoint metadata key
+ * - Checkpoint metadata key (optional — when omitted, the validator auto-resolves the
+ * active streamer key from commit metadata using
+ * {@link org.apache.hudi.common.table.checkpoint.CheckpointUtils#getCheckpoint(HoodieCommitMetadata)},
+ * which prefers V2 and falls back to V1. Subclasses that read a custom non-streamer key
+ * (e.g. Flink's HOODIE_METADATA_KEY) must pass it explicitly.)
* - Source-specific parsing logic (if needed)
*
* Configuration:
@@ -66,7 +72,26 @@ public abstract class StreamingOffsetValidator extends BasePreCommitValidator {
protected final CheckpointFormat checkpointFormat;
/**
- * Create a streaming offset validator.
+ * Create a streaming offset validator that auto-resolves the checkpoint key from commit
+ * metadata using {@link org.apache.hudi.common.table.checkpoint.CheckpointUtils#getCheckpoint(HoodieCommitMetadata)}.
+ *
+ * Use this constructor for streamer pipelines (V1 or V2 checkpoint keys). The validator
+ * will prefer V2 (table version 8+) and fall back to V1 transparently, so subclasses don't
+ * need to know which key the writer used.
+ *
+ * @param config Validator configuration
+ * @param checkpointFormat Format of the checkpoint string
+ */
+ protected StreamingOffsetValidator(TypedProperties config,
+ CheckpointFormat checkpointFormat) {
+ this(config, null, checkpointFormat);
+ }
+
+ /**
+ * Create a streaming offset validator with an explicit checkpoint metadata key.
+ *
+ * Use this constructor when the writer stores its checkpoint under a custom key that
+ * is not the standard streamer V1/V2 key (e.g. Flink's HOODIE_METADATA_KEY).
*
* @param config Validator configuration
* @param checkpointKey Key to extract checkpoint from extraMetadata
@@ -95,10 +120,12 @@ public void validateWithMetadata(ValidationContext context) throws HoodieValidat
return;
}
- // Extract current checkpoint
- Option currentCheckpointOpt = context.getExtraMetadata(checkpointKey);
+ // Extract current checkpoint — either from the explicit key (custom writers like Flink) or
+ // by auto-resolving from commit metadata (streamer pipelines, V2-then-V1 fallback).
+ Option currentCheckpointOpt = resolveCheckpoint(context.getCommitMetadata());
if (!currentCheckpointOpt.isPresent()) {
- log.warn("Current checkpoint not found with key: {}. Skipping validation.", checkpointKey);
+ log.warn("Current checkpoint not found (key: {}). Skipping validation.",
+ checkpointKey == null ? "" : checkpointKey);
return;
}
String currentCheckpoint = currentCheckpointOpt.get();
@@ -110,8 +137,7 @@ public void validateWithMetadata(ValidationContext context) throws HoodieValidat
}
// Extract previous checkpoint
- Option previousCheckpointOpt = context.getPreviousCommitMetadata()
- .flatMap(metadata -> Option.ofNullable(metadata.getMetadata(checkpointKey)));
+ Option previousCheckpointOpt = resolveCheckpoint(context.getPreviousCommitMetadata());
if (!previousCheckpointOpt.isPresent()) {
log.info("Previous checkpoint not found. May be first streaming commit. Skipping validation.");
@@ -139,6 +165,10 @@ public void validateWithMetadata(ValidationContext context) throws HoodieValidat
long recordsWritten = context.getTotalInsertRecordsWritten()
+ context.getTotalUpdateRecordsWritten();
+ // Track write errors so callers can distinguish write-failure deviation (write errors > 0)
+ // from silent data loss (write errors == 0) when the validator fires.
+ long writeErrors = context.getTotalWriteErrors();
+
// For empty commits (e.g., no new data from source), both offsetDiff and recordsWritten
// can be zero. This is a valid scenario — skip validation to avoid false positives.
if (offsetDifference == 0 && recordsWritten == 0) {
@@ -147,7 +177,7 @@ public void validateWithMetadata(ValidationContext context) throws HoodieValidat
}
// Validate offset vs record consistency
- validateOffsetConsistency(offsetDifference, recordsWritten,
+ validateOffsetConsistency(offsetDifference, recordsWritten, writeErrors,
currentCheckpoint, previousCheckpoint);
}
@@ -155,12 +185,13 @@ public void validateWithMetadata(ValidationContext context) throws HoodieValidat
* Validate that offset difference matches record count within tolerance.
*
* @param offsetDiff Expected records based on offset difference
- * @param recordsWritten Actual records written
+ * @param recordsWritten Actual records written (inserts + updates)
+ * @param writeErrors Records that failed to write (tracked in write status errors)
* @param currentCheckpoint Current checkpoint string (for error messages)
* @param previousCheckpoint Previous checkpoint string (for error messages)
* @throws HoodieValidationException if validation fails and policy is FAIL
*/
- protected void validateOffsetConsistency(long offsetDiff, long recordsWritten,
+ protected void validateOffsetConsistency(long offsetDiff, long recordsWritten, long writeErrors,
String currentCheckpoint, String previousCheckpoint)
throws HoodieValidationException {
@@ -169,20 +200,23 @@ protected void validateOffsetConsistency(long offsetDiff, long recordsWritten,
if (deviation > tolerancePercentage) {
String errorMsg = String.format(
"Streaming offset validation failed. "
- + "Offset difference: %d, Records written: %d, Deviation: %.2f%%, Tolerance: %.2f%%. "
- + "This may indicate data loss or filtering. "
+ + "Offset difference: %d, Records written: %d, Write errors: %d, Deviation: %.2f%%, Tolerance: %.2f%%. "
+ + "%s"
+ "Previous checkpoint: %s, Current checkpoint: %s",
- offsetDiff, recordsWritten, deviation, tolerancePercentage,
+ offsetDiff, recordsWritten, writeErrors, deviation, tolerancePercentage,
+ writeErrors > 0
+ ? "Non-zero write errors suggest records failed to write rather than silent data loss. "
+ : "This may indicate data loss or filtering. ",
previousCheckpoint, currentCheckpoint);
if (failurePolicy == ValidationFailurePolicy.WARN_LOG) {
- log.warn(errorMsg + " (failure policy is WARN_LOG, commit will proceed)");
+ log.warn("{} (failure policy is WARN_LOG, commit will proceed)", errorMsg);
} else {
throw new HoodieValidationException(errorMsg);
}
} else {
- log.info("Offset validation passed. Offset diff: {}, Records: {}, Deviation: {}% (within {}%)",
- offsetDiff, recordsWritten, String.format("%.2f", deviation), tolerancePercentage);
+ log.info("Offset validation passed. Offset diff: {}, Records: {}, Write errors: {}, Deviation: {}% (within {}%)",
+ offsetDiff, recordsWritten, writeErrors, String.format("%.2f", deviation), tolerancePercentage);
}
}
@@ -210,4 +244,33 @@ private double calculateDeviation(long offsetDiff, long recordsWritten) {
long difference = Math.abs(offsetDiff - recordsWritten);
return (100.0 * difference) / offsetDiff;
}
+
+ /**
+ * Resolve the checkpoint string from commit metadata.
+ *
+ * When the validator was constructed with an explicit {@code checkpointKey}, that key
+ * is read directly. Otherwise, {@link org.apache.hudi.common.table.checkpoint.CheckpointUtils#getCheckpoint(HoodieCommitMetadata)}
+ * is used to locate the active streamer checkpoint (V2 first, V1 fallback), so callers
+ * don't need to know which key the writer used.
+ *
+ * @param commitMetadataOpt Optional commit metadata containing extraMetadata
+ * @return Optional checkpoint string (empty if metadata is absent or no checkpoint key matches)
+ */
+ private Option resolveCheckpoint(Option commitMetadataOpt) {
+ if (!commitMetadataOpt.isPresent()) {
+ return Option.empty();
+ }
+ HoodieCommitMetadata metadata = commitMetadataOpt.get();
+ if (checkpointKey != null) {
+ return Option.ofNullable(metadata.getMetadata(checkpointKey));
+ }
+ try {
+ return Option.ofNullable(
+ org.apache.hudi.common.table.checkpoint.CheckpointUtils.getCheckpoint(metadata)
+ .getCheckpointKey());
+ } catch (HoodieException e) {
+ // No V1 or V2 streamer checkpoint key present in extraMetadata.
+ return Option.empty();
+ }
+ }
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieArchivalConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieArchivalConfig.java
index 8854c87edeaba..e97e268fa9f9f 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieArchivalConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieArchivalConfig.java
@@ -88,6 +88,15 @@ public class HoodieArchivalConfig extends HoodieConfig {
.withDocumentation("Archiving of instants is batched in best-effort manner, to pack more instants into a single"
+ " archive log. This config controls such archival batch size.");
+ public static final ConfigProperty MIGRATION_COMMITS_ARCHIVAL_BATCH_SIZE = ConfigProperty
+ .key("hoodie.timeline.migration.commits.archival.batch")
+ .defaultValue(500)
+ .markAdvanced()
+ .withDocumentation("Batch size used when migrating the legacy archived timeline to the LSM timeline during a"
+ + " table version upgrade. A larger batch size minimizes the number of parquet files (and the associated"
+ + " remote storage operations like exists check, parquet write and manifest update) created during the"
+ + " one-time migration, which significantly reduces the total migration time.");
+
public static final ConfigProperty TIMELINE_COMPACTION_BATCH_SIZE = ConfigProperty
.key("hoodie.timeline.compaction.batch.size")
.defaultValue(10)
@@ -211,6 +220,11 @@ public HoodieArchivalConfig.Builder withCommitsArchivalBatchSize(int batchSize)
return this;
}
+ public HoodieArchivalConfig.Builder withMigrationCommitsArchivalBatchSize(int batchSize) {
+ archivalConfig.setValue(MIGRATION_COMMITS_ARCHIVAL_BATCH_SIZE, String.valueOf(batchSize));
+ return this;
+ }
+
public Builder withArchiveBeyondSavepoint(boolean archiveBeyondSavepoint) {
archivalConfig.setValue(ARCHIVE_BEYOND_SAVEPOINT, String.valueOf(archiveBeyondSavepoint));
return this;
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java
index 9cedac1be74d0..1d2c6bb2b426b 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java
@@ -29,6 +29,7 @@
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.index.bucket.partition.PartitionBucketIndexRule;
+import org.apache.hudi.keygen.KeyGenUtils;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
import lombok.Getter;
@@ -39,9 +40,8 @@
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
-import java.util.Arrays;
+import java.util.List;
import java.util.Properties;
-import java.util.stream.Collectors;
import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_DYNAMIC_MAX_ENTRIES;
import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_FPP_VALUE;
@@ -777,10 +777,9 @@ private void validateBucketIndexConfig() {
hoodieIndexConfig.setValue(BUCKET_INDEX_HASH_FIELD,
hoodieIndexConfig.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME));
} else {
- boolean valid = Arrays
- .stream(hoodieIndexConfig.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME).split(","))
- .collect(Collectors.toSet())
- .containsAll(Arrays.asList(hoodieIndexConfig.getString(BUCKET_INDEX_HASH_FIELD).split(",")));
+ List recordKeyFields = KeyGenUtils.getRecordKeyFields(hoodieIndexConfig.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME));
+ List indexKeyFields = KeyGenUtils.getIndexKeyFields(hoodieIndexConfig.getString(BUCKET_INDEX_HASH_FIELD));
+ boolean valid = recordKeyFields.containsAll(indexKeyFields);
if (!valid) {
throw new HoodieIndexException("Bucket index key (if configured) must be subset of record key.");
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodiePreCommitValidatorConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodiePreCommitValidatorConfig.java
index f85cc44120d4e..f4999bc39e166 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodiePreCommitValidatorConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodiePreCommitValidatorConfig.java
@@ -43,7 +43,12 @@ public class HoodiePreCommitValidatorConfig extends HoodieConfig {
.key("hoodie.precommit.validators")
.defaultValue("")
.markAdvanced()
- .withDocumentation("Comma separated list of class names that can be invoked to validate commit");
+ .withDocumentation("Comma separated list of class names that can be invoked to validate commit. "
+ + "Available streaming offset validators: "
+ + "org.apache.hudi.sink.validator.FlinkKafkaOffsetValidator (Flink Kafka), "
+ + "org.apache.hudi.utilities.streamer.validator.SparkKafkaOffsetValidator (Spark/HoodieStreamer Kafka). "
+ + "Available write-error validators: "
+ + "org.apache.hudi.utilities.streamer.validator.SparkWriteErrorValidator (Spark/HoodieStreamer write errors).");
public static final String VALIDATOR_TABLE_VARIABLE = "";
public static final ConfigProperty EQUALITY_SQL_QUERIES = ConfigProperty
@@ -71,7 +76,8 @@ public class HoodiePreCommitValidatorConfig extends HoodieConfig {
.markAdvanced()
.withDocumentation("Tolerance percentage for streaming offset validation "
+ "(used by org.apache.hudi.client.validator.StreamingOffsetValidator "
- + "and org.apache.hudi.sink.validator.FlinkKafkaOffsetValidator). "
+ + "and org.apache.hudi.sink.validator.FlinkKafkaOffsetValidator "
+ + "and org.apache.hudi.utilities.streamer.validator.SparkKafkaOffsetValidator). "
+ "The validator compares the offset difference (expected records from source) "
+ "with actual records written. If the deviation exceeds this percentage, "
+ "the commit is rejected or warned depending on the validation failure policy. "
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
index 5df834121bf90..f5cc2bf67a3e3 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
@@ -74,6 +74,7 @@
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.execution.bulkinsert.BulkInsertSortMode;
import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.internal.schema.utils.SchemaChangeUtils;
import org.apache.hudi.io.FileGroupReaderBasedMergeHandle;
import org.apache.hudi.io.HoodieConcatHandle;
import org.apache.hudi.keygen.SimpleAvroKeyGenerator;
@@ -680,9 +681,11 @@ public class HoodieWriteConfig extends HoodieConfig {
public static final ConfigProperty CLIENT_HEARTBEAT_NUM_TOLERABLE_MISSES = ConfigProperty
.key("hoodie.client.heartbeat.tolerable.misses")
- .defaultValue(2)
+ .defaultValue(10)
.markAdvanced()
- .withDocumentation("Number of heartbeat misses, before a writer is deemed not alive and all pending writes are aborted.");
+ .withDocumentation("Number of heartbeat misses, before a writer is deemed not alive and all pending writes are aborted. "
+ + "A higher value tolerates transient driver pauses (e.g. GC) or storage-latency spikes that would otherwise "
+ + "delay a heartbeat and cause a still-healthy writer's commit to be aborted.");
public static final ConfigProperty CLUSTERING_BLOCK_FOR_PENDING_INGESTION = ConfigProperty
.key("hoodie.clustering.fail.on.pending.ingestion.during.conflict.resolution")
@@ -768,21 +771,22 @@ public class HoodieWriteConfig extends HoodieConfig {
.markAdvanced()
.sinceVersion("1.2.0")
.withDocumentation("Comma-separated list of extra metadata keys that should be automatically carried forward "
- + "to every new commit. These keys will be read from recent commit metadata and included in new commits, "
- + "ensuring they remain accessible without walking the timeline or worrying about archival. "
- + "This is useful for tracking checkpoint information (e.g., Kafka offsets, Flink checkpoints) or any metadata "
- + "that needs to persist across commits. New values override old ones. Only applies to data table commits.");
+ + "to every new commit and clean instant. These keys will be read from recent commit and clean metadata "
+ + "and included in new commits/cleans, ensuring they remain accessible without walking the timeline or "
+ + "worrying about archival. This is useful for tracking checkpoint information (e.g., Kafka offsets, "
+ + "Flink checkpoints) or any metadata that needs to persist across commits. New values override old ones. "
+ + "Only applies to data table commits and clean instants.");
public static final ConfigProperty ROLLING_METADATA_TIMELINE_LOOKBACK_COMMITS = ConfigProperty
.key("hoodie.write.rolling.metadata.timeline.lookback.commits")
.defaultValue(10)
.markAdvanced()
.sinceVersion("1.2.0")
- .withDocumentation("Maximum number of completed commits to walk back in the timeline when searching for "
- + "rolling metadata keys. If a rolling metadata key is not found in the latest commit, the system will "
- + "walk back up to this many commits to find the most recent value. This ensures rolling metadata is "
- + "preserved even if some commits don't update all keys. Higher values provide more resilience but may "
- + "impact performance. Only applies when hoodie.write.rolling.metadata.keys is configured.");
+ .withDocumentation("Maximum number of completed instants (commits and clean) to walk back in the timeline "
+ + "when searching for rolling metadata keys. If a rolling metadata key is not found in the latest instant, "
+ + "the system will walk back up to this many instants to find the most recent value. This ensures rolling "
+ + "metadata is preserved even if some instants don't carry all keys. Higher values provide more resilience "
+ + "but may impact performance. Only applies when hoodie.write.rolling.metadata.keys is configured.");
public static final ConfigProperty ALLOW_OPERATION_METADATA_FIELD = ConfigProperty
.key("hoodie.allow.operation.metadata.field")
@@ -2018,6 +2022,10 @@ public int getCommitArchivalBatchSize() {
return getInt(HoodieArchivalConfig.COMMITS_ARCHIVAL_BATCH_SIZE);
}
+ public int getMigrationCommitArchivalBatchSize() {
+ return getInt(HoodieArchivalConfig.MIGRATION_COMMITS_ARCHIVAL_BATCH_SIZE);
+ }
+
public boolean shouldBlockArchivalOnCleanECTR() {
return getBoolean(HoodieArchivalConfig.BLOCK_ARCHIVAL_ON_LATEST_CLEAN_ECTR);
}
@@ -3857,6 +3865,11 @@ private void validate() {
+ "schedule inline compaction (%s) can be enabled. Both can't be set to true at the same time. %s, %s", HoodieCompactionConfig.INLINE_COMPACT.key(),
HoodieCompactionConfig.SCHEDULE_INLINE_COMPACT.key(), inlineCompact, inlineCompactSchedule));
+ // Parse-and-discard so a malformed 'field:type' entry fails at client build time rather
+ // than deep inside deduceWriterSchema on the first commit. Empty (default) is a no-op.
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(
+ writeConfig.getStringOrDefault(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES));
+
int lookbackCommits = writeConfig.getInt(ROLLING_METADATA_TIMELINE_LOOKBACK_COMMITS);
checkArgument(lookbackCommits >= 0,
String.format("%s must be non-negative, but was %d",
@@ -3890,7 +3903,9 @@ private String getDefaultMarkersType(EngineType engineType) {
}
case FLINK:
case JAVA:
- // Timeline-server-based marker is not supported for Flink and Java engines
+ // Timeline-server-based markers are not the default for Flink and Java, but they are not
+ // unsupported either: setting hoodie.write.markers.type explicitly selects them, subject to the
+ // same gates WriteMarkersFactory applies to every engine.
return MarkerType.DIRECT.toString();
default:
throw new HoodieNotSupportedException("Unsupported engine " + engineType);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/FileMetadataWriteStatusConverter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/FileMetadataWriteStatusConverter.java
index 51ac55cd5dc4d..8e96b28adc3b0 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/FileMetadataWriteStatusConverter.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/FileMetadataWriteStatusConverter.java
@@ -60,7 +60,7 @@ public FileMetadataWriteStatusConverter(HoodieTable hoodieTable, Hoo
*/
public WriteStatus convert(String parquetFile, String partitionPath,
Map executionConfigs) throws IOException {
- LOG.info("Creating write status for parquet file " + parquetFile);
+ LOG.info("Creating write status for parquet file {}", parquetFile);
WriteStatus writeStatus = (WriteStatus) ReflectionUtils.loadClass(this.writeConfig.getWriteStatusClassName(),
this.hoodieTable.shouldTrackSuccessRecords(), this.writeConfig.getWriteStatusFailureFraction(), this.hoodieTable.isMetadataTable());
StoragePath parquetFilePath = new StoragePath(parquetFile);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
index d9fe1068e4218..d2cc530295f2b 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
@@ -319,14 +319,16 @@ private static HoodieData> getExistingRecords(
Option internalSchemaOption = SerDeHelper.fromJson(config.getInternalSchema());
FileSlice fileSlice = fileSliceOption.get();
HoodieReaderContext readerContext = readerContextFactory.getContext();
- HoodieFileGroupReader fileGroupReader = HoodieFileGroupReader.newBuilder()
+ HoodieFileGroupReader fileGroupReader = HoodieFileGroupReader.builder()
.withReaderContext(readerContext)
.withHoodieTableMetaClient(metaClient)
.withLatestCommitTime(instantTime.get())
- .withFileSlice(fileSlice)
+ .withBaseFileOption(fileSlice.getBaseFile())
+ .withLogFiles(fileSlice.getLogFiles())
+ .withPartitionPath(fileSlice.getPartitionPath())
.withDataSchema(dataSchema)
.withRequestedSchema(dataSchema)
- .withInternalSchema(internalSchemaOption)
+ .withInternalSchemaOpt(internalSchemaOption)
.withProps(metaClient.getTableConfig().getProps())
.build();
try {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
index eed3ab39599c1..2bde3aec815b4 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/BucketIdentifier.java
@@ -42,7 +42,7 @@ public static int getBucketId(List hashKeyFields, int numBuckets) {
}
protected static List getHashKeys(String recordKey, String indexKeyFields) {
- return getHashKeysUsingIndexFields(recordKey, Arrays.asList(indexKeyFields.split(",")));
+ return getHashKeysUsingIndexFields(recordKey, KeyGenUtils.getIndexKeyFields(indexKeyFields));
}
protected static List getHashKeys(String recordKey, List indexKeyFields) {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/ConsistentBucketIndexUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/ConsistentBucketIndexUtils.java
index 5d02de2cbcfd3..b522a77af83c6 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/ConsistentBucketIndexUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/ConsistentBucketIndexUtils.java
@@ -175,7 +175,7 @@ public static Option loadMetadata(HoodieTable t
} catch (FileNotFoundException e) {
return Option.empty();
} catch (IOException e) {
- log.error("Error when loading hashing metadata, partition: " + partition, e);
+ log.error("Error when loading hashing metadata, partition: {}", partition, e);
throw new HoodieIndexException("Error while loading hashing metadata", e);
}
}
@@ -258,7 +258,7 @@ private static Option loadMetadataFromGivenFile
} catch (FileNotFoundException e) {
return Option.empty();
} catch (IOException e) {
- log.error("Error when loading hashing metadata, for path: " + metaFile.getPath().getName(), e);
+ log.error("Error when loading hashing metadata, for path: {}", metaFile.getPath().getName(), e);
throw new HoodieIndexException("Error while loading hashing metadata", e);
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java
index 38c7cb5319a3f..be61454fae36c 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bucket/HoodieBucketIndex.java
@@ -29,13 +29,13 @@
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieIndexException;
import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.keygen.KeyGenUtils;
import org.apache.hudi.table.HoodieTable;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import java.io.Serializable;
-import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -57,8 +57,8 @@ public HoodieBucketIndex(HoodieWriteConfig config) {
super(config);
this.numBuckets = config.getBucketIndexNumBuckets();
- this.indexKeyFields = Arrays.asList(config.getBucketIndexHashField().split(","));
- log.info("Use bucket index, numBuckets = " + numBuckets + ", indexFields: " + indexKeyFields);
+ this.indexKeyFields = KeyGenUtils.getIndexKeyFields(config.getBucketIndexHashField());
+ log.info("Use bucket index, numBuckets = {}, indexFields: {}", numBuckets, indexKeyFields);
}
@Override
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
index 8144ae4c2f859..6eb999d74b550 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
@@ -30,6 +30,7 @@
import org.apache.hudi.common.model.MetadataValues;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieInsertException;
@@ -117,7 +118,7 @@ protected void doWrite(HoodieRecord record, HoodieSchema schema, TypedProperties
// record successful.
record.deflate();
} catch (Throwable t) {
- log.error("Error writing record " + record, t);
+ log.error("Error writing record {}", record, t);
if (!config.getIgnoreWriteFailed()) {
throw new HoodieException(t.getMessage(), t);
}
@@ -131,8 +132,10 @@ protected void doWrite(HoodieRecord record, HoodieSchema schema, TypedProperties
public void write() {
Iterator keyIterator;
if (hoodieTable.requireSortedRecords()) {
- // Sorting the keys limits the amount of extra memory required for writing sorted records
- keyIterator = recordMap.keySet().stream().sorted().iterator();
+ // Sorting the keys limits the amount of extra memory required for writing sorted records.
+ // requireSortedRecords() is true only for HFile base files, which order keys by UTF-8 bytes,
+ // not String (UTF-16) order, so sort with the matching comparator.
+ keyIterator = recordMap.keySet().stream().sorted(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR).iterator();
} else {
keyIterator = recordMap.keySet().stream().iterator();
}
@@ -178,7 +181,7 @@ public IOType getIOType() {
*/
@Override
public List close() {
- log.info("Closing the file " + writeStatus.getFileId() + " as we are done with all the records " + recordsWritten);
+ log.info("Closing the file {} as we are done with all the records {}", writeStatus.getFileId(), recordsWritten);
try {
if (isClosed()) {
// Handle has already been closed
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/ExternalFileClusteringWriteHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/ExternalFileClusteringWriteHandle.java
index 9c9a5a3f0ba1b..f947ff9bc2b96 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/ExternalFileClusteringWriteHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/ExternalFileClusteringWriteHandle.java
@@ -62,7 +62,7 @@ public ExternalFileClusteringWriteHandle(HoodieWriteConfig config, String instan
// Create inProgress marker file
createMarkerFile(partitionPath, path.getName());
- LOG.info("New ExternalFileClusteringWriteHandle for partition :" + partitionPath + " with fileId " + fileId);
+ LOG.info("New ExternalFileClusteringWriteHandle for partition :{} with fileId {}", partitionPath, fileId);
}
/**
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedAppendHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedAppendHandle.java
index a081709f6fc22..c40ce0158a3cd 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedAppendHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedAppendHandle.java
@@ -82,10 +82,20 @@ public void doAppend() {
new HoodieLogFile(new StoragePath(FSUtils.constructAbsolutePath(
config.getBasePath(), operation.getPartitionPath()), logFileName)));
// Initializes the record iterator, log compaction requires writing the deletes into the delete block of the resulting log file.
- try (HoodieFileGroupReader fileGroupReader = HoodieFileGroupReader.newBuilder().withReaderContext(readerContext).withHoodieTableMetaClient(hoodieTable.getMetaClient())
- .withLatestCommitTime(instantTime).withPartitionPath(partitionPath).withLogFiles(logFiles).withBaseFileOption(Option.empty()).withDataSchema(writeSchemaWithMetaFields)
- .withRequestedSchema(writeSchemaWithMetaFields).withInternalSchema(internalSchemaOption).withProps(props).withEmitDelete(true)
- .withShouldUseRecordPosition(usePosition).withSortOutput(hoodieTable.requireSortedRecords())
+ try (HoodieFileGroupReader fileGroupReader = HoodieFileGroupReader.builder()
+ .withReaderContext(readerContext)
+ .withHoodieTableMetaClient(hoodieTable.getMetaClient())
+ .withLatestCommitTime(instantTime)
+ .withPartitionPath(partitionPath)
+ .withLogFiles(logFiles)
+ .withBaseFileOption(Option.empty())
+ .withDataSchema(writeSchemaWithMetaFields)
+ .withRequestedSchema(writeSchemaWithMetaFields)
+ .withInternalSchemaOpt(internalSchemaOption)
+ .withProps(props)
+ .withEmitDelete(true)
+ .withShouldUseRecordPosition(usePosition)
+ .withSortOutput(hoodieTable.requireSortedRecords())
// instead of using config.enableOptimizedLogBlocksScan(), we set to true as log compaction blocks only supported in scanV2
.build()) {
recordItr = new CloseableMappingIterator<>(fileGroupReader.getLogRecordsOnly(), record -> {
@@ -96,7 +106,7 @@ public void doAppend() {
header.put(HoodieLogBlock.HeaderMetadataType.COMPACTED_BLOCK_TIMES,
StringUtils.join(fileGroupReader.getValidBlockInstants(), ","));
super.doAppend();
- this.readStats = fileGroupReader.getStats();
+ this.readStats = fileGroupReader.getReadStats();
} catch (IOException e) {
throw new HoodieIOException("Failed to initialize file group reader for " + fileId, e);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
index d708c15f33845..b9432b626cd51 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/FileGroupReaderBasedMergeHandle.java
@@ -51,6 +51,7 @@
import org.apache.hudi.exception.HoodieUpsertException;
import org.apache.hudi.internal.schema.InternalSchema;
import org.apache.hudi.internal.schema.utils.AvroSchemaEvolutionUtils;
+import org.apache.hudi.internal.schema.utils.SchemaChangeUtils;
import org.apache.hudi.internal.schema.utils.SerDeHelper;
import org.apache.hudi.io.storage.HoodieFileWriterFactory;
import org.apache.hudi.keygen.BaseKeyGenerator;
@@ -257,8 +258,10 @@ public void doMerge() {
}
boolean usePosition = config.getBooleanOrDefault(MERGE_USE_RECORD_POSITIONS);
Option internalSchemaOption = SerDeHelper.fromJson(config.getInternalSchema())
- .map(internalSchema -> AvroSchemaEvolutionUtils.reconcileSchema(writeSchemaWithMetaFields.toAvroSchema(), internalSchema,
- config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS)));
+ .map(internalSchema -> AvroSchemaEvolutionUtils.reconcileSchema(writeSchemaWithMetaFields, internalSchema,
+ config.getBooleanOrDefault(HoodieCommonConfig.SET_NULL_FOR_MISSING_COLUMNS),
+ SchemaChangeUtils.parseTimestampLogicalTypeOverrides(
+ config.getStringOrDefault(HoodieCommonConfig.TIMESTAMP_LOGICAL_TYPE_OVERRIDES))));
long maxMemoryPerCompaction = getMaxMemoryForMerge();
props.put(HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE.key(), String.valueOf(maxMemoryPerCompaction));
Option> logFilesStreamOpt = compactionOperation.map(op -> op.getDeltaFileNames().stream().map(logFileName ->
@@ -301,7 +304,7 @@ public void doMerge() {
// The stats of inserts, updates, and deletes are updated once at the end
// These will be set in the write stat when closing the merge handle
- this.readStats = fileGroupReader.getStats();
+ this.readStats = fileGroupReader.getReadStats();
this.insertRecordsWritten = readStats.getNumInserts();
this.updatedRecordsWritten = readStats.getNumUpdates();
this.recordsDeleted = readStats.getNumDeletes();
@@ -318,10 +321,10 @@ protected long getMaxMemoryForMerge() {
private HoodieFileGroupReader getFileGroupReader(boolean usePosition, Option internalSchemaOption, TypedProperties props,
Option> logFileStreamOpt, Iterator> incomingRecordsItr) {
- HoodieFileGroupReader.Builder fileGroupBuilder = HoodieFileGroupReader.newBuilder().withReaderContext(readerContext).withHoodieTableMetaClient(hoodieTable.getMetaClient())
+ HoodieFileGroupReader.HoodieFileGroupReaderBuilder fileGroupBuilder = HoodieFileGroupReader.builder().withReaderContext(readerContext).withHoodieTableMetaClient(hoodieTable.getMetaClient())
.withLatestCommitTime(maxInstantTime).withPartitionPath(partitionPath).withBaseFileOption(Option.ofNullable(baseFileToMerge))
.withDataSchema(writeSchemaWithMetaFields).withRequestedSchema(writeSchemaWithMetaFields)
- .withInternalSchema(internalSchemaOption).withProps(props)
+ .withInternalSchemaOpt(internalSchemaOption).withProps(props)
.withShouldUseRecordPosition(usePosition).withSortOutput(hoodieTable.requireSortedRecords())
.withFileGroupUpdateCallback(createCallback());
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
index 5ea8ba460f873..e76d3dbd77bcd 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
@@ -39,7 +39,7 @@
import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.AppendResult;
-import org.apache.hudi.common.table.log.HoodieLogFormat.Writer;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
import org.apache.hudi.common.table.log.block.HoodieHFileDataBlock;
@@ -54,6 +54,7 @@
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.common.util.SizeEstimator;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieAppendException;
@@ -105,7 +106,7 @@ public class HoodieAppendHandle extends HoodieWriteHandle> recordItr;
// Writer to log into the file group's latest slice.
- protected Writer writer;
+ protected HoodieLogFormat.Writer writer;
protected final List statuses;
// Total number of records written during appending
@@ -260,7 +261,7 @@ private void init(HoodieRecord record) {
? getInstantTimeForLogFile(record) : deltaWriteStat.getPrevCommit();
this.writer = createLogWriter(instantTime, fileSliceOpt);
} catch (Exception e) {
- log.error("Error in update task at commit " + instantTime, e);
+ log.error("Error in update task at commit {}", instantTime, e);
writeStatus.setGlobalError(e);
throw new HoodieUpsertException("Failed to initialize HoodieAppendHandle for FileId: " + fileId + " on commit "
+ instantTime + " on storage path " + hoodieTable.getMetaClient().getBasePath() + "/" + partitionPath, e);
@@ -558,14 +559,16 @@ public List close() {
writer = null;
}
- // update final size, once for all log files
- // TODO we can actually deduce file size purely from AppendResult (based on offset and size
- // of the appended block)
+ // Set the final on-disk size of each log file. Appends within an append handle are contiguous,
+ // so a log file's length equals its start offset plus the total bytes appended to it. That is
+ // exactly what fs.getFileStatus().getLength() returns, and both values are already captured by
+ // the AppendResult stats (logOffset and the accumulated fileSizeInBytes). Deriving the size this
+ // way avoids a getPathInfo/HEAD per log file, which is a remote round trip per file group on
+ // object stores.
for (WriteStatus status : statuses) {
- long logFileSize = storage.getPathInfo(
- new StoragePath(config.getBasePath(), status.getStat().getPath()))
- .getLength();
- status.getStat().setFileSizeInBytes(logFileSize);
+ HoodieDeltaWriteStat stat = (HoodieDeltaWriteStat) status.getStat();
+ long appendedBytes = stat.getFileSizeInBytes();
+ stat.setFileSizeInBytes(stat.getLogOffset() + appendedBytes);
}
// generate Secondary index stats if streaming writes is enabled.
@@ -725,7 +728,9 @@ protected HoodieLogBlock getDataBlock(HoodieWriteConfig writeConfig,
case HFILE_DATA_BLOCK:
// Not supporting positions in HFile data blocks
header.remove(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS);
- records.sort(Comparator.comparing(HoodieRecord::getRecordKey));
+ // HFile orders keys by their raw UTF-8 bytes, so sort by UTF-8 bytes rather than
+ // String (UTF-16) order to keep non-ASCII / binary keys consistent with the writer.
+ records.sort(Comparator.comparing(HoodieRecord::getRecordKey, StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR));
return new HoodieHFileDataBlock(
records, header, writeConfig.getHFileCompressionAlgorithm(), new StoragePath(writeConfig.getBasePath()));
case PARQUET_DATA_BLOCK:
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java
index 94a86f1f94762..fa7baab2af150 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java
@@ -70,7 +70,7 @@ private MessageType getWriteSchema(HoodieWriteConfig config, List i
try {
ParquetUtils parquetUtils = new ParquetUtils();
MessageType fileSchema = parquetUtils.readMessageType(table.getStorage(), inputFiles.get(0));
- log.info("Binary copy schema evolution disabled. Using schema from input file: " + inputFiles.get(0));
+ log.info("Binary copy schema evolution disabled. Using schema from input file: {}", inputFiles.get(0));
return fileSchema;
} catch (Exception e) {
log.error("Failed to read schema from input file", e);
@@ -109,8 +109,8 @@ public HoodieBinaryCopyHandle(
}
public void write() {
- log.info("Start to merge source files " + this.inputFiles + " into target file: " + this.path
- + ". Please pay attention that we will not rolling files based on max-file-size config during binary copy.");
+ log.info("Start to merge source files {} into target file: {}. Please pay attention that we will not rolling files based on max-file-size config during binary copy.",
+ this.inputFiles, this.path);
HoodieTimer timer = HoodieTimer.start();
long records = 0;
try {
@@ -123,12 +123,12 @@ public void write() {
this.recordsWritten = records;
this.insertRecordsWritten = records;
}
- log.info("Finish rewriting " + this.path + ". Using " + timer.endTimer() + " mills");
+ log.info("Finish rewriting {}. Using {} mills", this.path, timer.endTimer());
}
@Override
public List close() {
- log.info("Closing the file " + writeStatus.getFileId() + " as we are done with all the records " + recordsWritten);
+ log.info("Closing the file {} as we are done with all the records {}", writeStatus.getFileId(), recordsWritten);
try {
this.writer.close();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
index 9456d5ce586bb..7cc74c40afeeb 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
@@ -24,6 +24,7 @@
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieUpsertException;
import org.apache.hudi.keygen.BaseKeyGenerator;
@@ -47,7 +48,7 @@
@NotThreadSafe
public class HoodieSortedMergeHandle extends HoodieWriteMergeHandle