This is an automated email from the ASF dual-hosted git repository.
nsivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 5cce63e83c48 feat(utilities): add metadata table sync and compaction
lag checks to HoodieTableHealthChecker (#20154)
5cce63e83c48 is described below
commit 5cce63e83c482e8c0dcd686919d00a1527b5ec9e
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Thu Oct 22 01:53:36 2026 -0700
feat(utilities): add metadata table sync and compaction lag checks to
HoodieTableHealthChecker (#20154)
* feat(utilities): add metadata table sync and compaction lag checks with
hudi-mdt-health skill
Two new checks, mdt-sync and mdt-compaction, plus the hudi-mdt-health skill
that interprets them.
mdt-sync compares the metadata table's file listing against storage for
the latest file slices
and reports which direction they diverge, since a rebuild only fixes one
of the two. It
recommends the validator with a scheme-qualified base path.
mdt-compaction measures delta commits since the metadata table's last
compaction against the
writer's configured trigger, and reports a metadata table that nothing
maintains instead of
telling the operator to wait out the lag.
Both report SKIPPED, naming the property, when the governing writer
configuration is not
supplied. Part of #19262.
---
hudi-ai-operator/skills/hudi-mdt-health/README.md | 65 ++++
hudi-ai-operator/skills/hudi-mdt-health/SKILL.md | 136 +++++++
.../skills/hudi-mdt-health/references/checks.md | 94 +++++
.../hudi/utilities/HoodieTableHealthChecker.java | 8 +-
.../main/java/org/apache/hudi/utilities/README.md | 53 +++
.../health/MetadataCompactionLagCheck.java | 231 +++++++++++
.../utilities/health/MetadataTablePresence.java | 51 +++
.../utilities/health/MetadataTableSyncCheck.java | 430 +++++++++++++++++++++
.../utilities/TestHoodieTableHealthChecker.java | 2 +-
.../health/TestMetadataCompactionLagCheck.java | 294 ++++++++++++++
.../health/TestMetadataTableSyncCheck.java | 165 ++++++++
.../TestMetadataTableSyncCheckFunctional.java | 253 ++++++++++++
12 files changed, 1779 insertions(+), 3 deletions(-)
diff --git a/hudi-ai-operator/skills/hudi-mdt-health/README.md
b/hudi-ai-operator/skills/hudi-mdt-health/README.md
new file mode 100644
index 000000000000..76135b6e5544
--- /dev/null
+++ b/hudi-ai-operator/skills/hudi-mdt-health/README.md
@@ -0,0 +1,65 @@
+<!--
+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.
+-->
+
+# Hudi Metadata Table Health
+
+A Claude Code Skill that runs the metadata-table checks of
`HoodieTableHealthChecker` against a
+Hudi table and turns the JSON report into a diagnosis: whether the metadata
table is caught up
+with the data table, whether its file listing can be trusted, whether its
compaction is keeping
+up, whether the record-level index is sized for the table, and what to do next.
+
+It pairs with the utility in `hudi-utilities`
+(`org.apache.hudi.utilities.HoodieTableHealthChecker`); see that class's `.md`
for the tool's own
+runbook. The Skill adds the interpretation layer — reading the report,
explaining findings, and
+recommending the next command — and is careful about the one recommendation
that is easy to get
+wrong: it never proposes rebuilding the metadata table unless a listing
inconsistency has been
+confirmed, because sync lag and compaction lag recover on their own and a
rebuild is disruptive.
+
+## Install
+
+```bash
+SKILL=hudi-ai-operator/skills/hudi-mdt-health
+
+# user-level
+mkdir -p ~/.claude/skills && cp -r "$SKILL" ~/.claude/skills/
+# or project-level
+mkdir -p .claude/skills && cp -r "$SKILL" .claude/skills/
+```
+
+Then `/hudi-mdt-health` in Claude Code. You will be asked for the table base
path and the writer
+properties file; have both ready.
+
+## Requirements
+
+- The `hudi-utilities-bundle` jar from the same Hudi release as this Skill,
reachable from wherever
+ Claude Code runs commands.
+- Read access to the table's base path, including `.hoodie/metadata`. The tool
is read-only and
+ never modifies the table or its metadata table.
+
+## Files
+
+- `SKILL.md` — the Skill itself.
+- `references/checks.md` — the JSON envelope, and per check: governing
properties, what
+ "unhealthy" means, and the remedy.
+
+## Scope
+
+This Skill covers the metadata table: sync with the data table, listing
consistency, compaction
+cadence, and record-index sizing. Data-table service cadence (compaction,
cleaning, archival,
+savepoints) and data layout (small files, micro-partitions) are separate tools
with their own
+Skills.
diff --git a/hudi-ai-operator/skills/hudi-mdt-health/SKILL.md
b/hudi-ai-operator/skills/hudi-mdt-health/SKILL.md
new file mode 100644
index 000000000000..47ac0e391c38
--- /dev/null
+++ b/hudi-ai-operator/skills/hudi-mdt-health/SKILL.md
@@ -0,0 +1,136 @@
+---
+name: hudi-mdt-health
+description: Checks whether an Apache Hudi table's metadata table is healthy
-- caught up with the data table, consistent with the files on storage,
compacting on schedule, and with a record-level index sized for the table -- by
running the HoodieTableHealthChecker utility's metadata-table checks and
interpreting its JSON report. Invoke when a user asks whether the metadata
table is in sync or lagging, whether file listings from the metadata table can
be trusted, why metadata-table compa [...]
+---
+
+<!--
+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.
+-->
+
+# Hudi Metadata Table Health
+
+You are helping an operator find out whether a Hudi table's metadata table can
be trusted and is
+keeping up, and what to do if it is not. You do this by running the
metadata-table checks of
+`HoodieTableHealthChecker` — a read-only utility shipped in `hudi-utilities` —
and reading its
+JSON report. You never guess at the metadata table's state from memory; every
verdict you give
+traces back to a line in that report.
+
+## What the tool answers
+
+| Check | Question |
+|---|---|
+| `mdt-sync` | Has every completed data-table write reached the metadata
table, and does its file listing match storage? |
+| `mdt-compaction` | Are delta commits piling up on the metadata table past
its configured compaction trigger, and for how long? |
+| `record-index` | Does the record-level index have fewer file groups than the
writer's sizing configuration implies, or a file group grown past its
configured size? |
+
+Each check reports `HEALTHY`, `UNHEALTHY`, or `SKIPPED`. `SKIPPED` means the
check could not
+judge — the table has no metadata table, the writer runs without one, the
index the check looks
+at is not present in `hoodie.table.metadata.partitions`, or a writer property
the check needs was
+not supplied. It is silence, not a pass.
+
+## Step 1 — collect what the tool needs
+
+Two things, and the second is the one users forget:
+
+1. **The table base path.**
+2. **The writer properties the table is written with.** The metadata-table
checks need
+ `hoodie.metadata.enable` at minimum; `mdt-compaction` also needs
+ `hoodie.metadata.compact.max.delta.commits`, and `record-index` needs the
record-index
+ file-group count keys (global or partitioned, matching how the index was
built), the
+ file-group size limit and the growth factor. The properties that matter per
check are listed
+ in `references/checks.md`. Ask for a properties file or the individual
`hoodie.*` values.
+
+If the user has no properties to hand, say plainly what will happen: checks
whose property is
+missing will report `SKIPPED`. Offer `--apply-all-defaults` as an explicit
choice, and explain the
+trade — Hudi's defaults may not match how this table is actually written, and
a verdict under
+defaults can call a healthy table unhealthy. **Never pass
`--apply-all-defaults` without the user
+choosing it.**
+
+## Step 2 — run it
+
+The tool needs the `hudi-utilities-bundle` jar on the classpath. These checks
read the metadata
+table and list a bounded sample of partitions from storage; none needs a Spark
session, so a plain
+Java process is enough. `spark-submit` also works.
+
+```bash
+java -cp "$HUDI_UTILITIES_BUNDLE"
org.apache.hudi.utilities.HoodieTableHealthChecker \
+ --base-path <table base path> \
+ --props <writer.properties> \
+ --checks mdt-sync,mdt-compaction,record-index \
+ --output JSON
+```
+
+Always request `--output JSON`. Capture the exit code as well as stdout: `0`
healthy, `1`
+unhealthy, `2` the tool could not run (bad arguments, unreadable table, or a
check that threw). On
+`2`, read the `errors` array and stderr before saying anything about the table.
+
+Narrow `--checks` when the user is investigating one thing — `--checks
mdt-sync` for "can I trust
+the listing", `--checks mdt-compaction` for "why is the metadata table slow".
+
+## Step 3 — read the report
+
+The JSON envelope is documented in `references/checks.md`. Dispatch on each
check's `status`:
+
+- **`UNHEALTHY`** — quote the `summary`, then the `findings` verbatim.
Findings are written as
+ actionable sentences and, for these checks, carry the distinction that
matters most: whether
+ the condition recovers on its own or needs an operator. Do not paraphrase
them into something
+ vaguer. Then add the remedy from `references/checks.md` for that check.
+- **`SKIPPED`** — the `summary` says why. If it names a missing property, tell
the user which to
+ supply. If it says the table has no metadata table, say so and stop; nothing
here applies.
+- **`HEALTHY`** — say so in one line. Do not pad.
+
+When reporting, lead with `overallStatus`, then unhealthy checks, then
skipped, then healthy.
+Include the `observed.*` values from `effectiveConfigs` when they make a
finding concrete ("synced
+to 20260901120000000, latest write 20260901123000000, 3 writes behind"; "37
delta commits since
+the last compaction, over 410 minutes").
+
+## Step 4 — recommend, don't act
+
+This skill diagnoses. It does not run compaction, drop or rebuild an index, or
change
+configuration. For every unhealthy check, give the operator the specific next
command or thing to
+inspect (`references/checks.md` has these), and say whether the action is
reversible. Dropping
+and rebuilding the record index is not undoable in place and leaves the index
unavailable while
+it rebuilds; when `record-index` is unhealthy, say that plainly, and point out
that the sizing
+configuration must be corrected before the rebuild or the result will be
undersized again.
+
+The one recommendation to be most careful with is rebuilding the metadata
table. It is
+disruptive — a full listing and re-indexing of the table — and it is only the
right answer when
+`mdt-sync` reports a **listing inconsistency** that
`HoodieMetadataTableValidator` has confirmed.
+Compaction lag does recover on its own once the next compaction runs, so
recommending a rebuild for
+it trades a transient condition for an outage.
+
+**Sync lag is different, and the distinction matters.** A later write does not
replay an instant the
+metadata table missed — the writer re-initializes only when the metadata
timeline is empty
+(`HoodieBackedTableMetadataWriter.isBootstrapNeeded`), so the next write
writes its own delta commit
+and moves the synced instant past the gap. The check then reads healthy while
the missed write's
+files stay absent from listings. So sync lag that does not correspond to an
in-flight write is a
+possible inconsistency, not a delay: confirm it with
`HoodieMetadataTableValidator`, which compares
+listings rather than instant times. Telling the user to wait is the one answer
that can leave the
+table silently wrong.
+
+## Guardrails
+
+- Every claim about the metadata table must come from the report. If you have
not run the tool,
+ say you have not, and do not speculate.
+- Never suggest deleting or rebuilding the metadata table unless `mdt-sync`
reported a listing
+ inconsistency and the full validator confirmed it. Lag on its own is not a
reason -- but do not
+ tell the user to wait for sync lag to clear either; run the validator and
act on what it finds.
+- Never pass `--apply-all-defaults` without the user's explicit choice.
+- Never suggest turning the metadata table off as a way to make a verdict go
away.
+- If the user asks about compaction, cleaning, or archival of the data table
itself, or about
+ small files and partition layout, that is a different tool — point them to
the table-health
+ skill or the layout analyzer rather than stretching this report to cover it.
diff --git a/hudi-ai-operator/skills/hudi-mdt-health/references/checks.md
b/hudi-ai-operator/skills/hudi-mdt-health/references/checks.md
new file mode 100644
index 000000000000..a96eceb5230b
--- /dev/null
+++ b/hudi-ai-operator/skills/hudi-mdt-health/references/checks.md
@@ -0,0 +1,94 @@
+<!--
+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.
+-->
+
+# Checks, properties, and remedies
+
+## JSON envelope
+
+```json
+{
+ "basePath": "...",
+ "tableType": "MERGE_ON_READ",
+ "overallStatus": "UNHEALTHY",
+ "checks": [
+ {
+ "name": "mdt-sync",
+ "status": "UNHEALTHY",
+ "summary": "one line for the operator",
+ "findings": ["actionable sentence", "..."],
+ "effectiveConfigs": {
+ "hoodie.metadata.enable": "true",
+ "effective.partitions.sampled.max": "20",
+ "observed.data.table.latest.completed.instant": "20260901123000000",
+ "observed.mdt.synced.instant": "20260901120000000",
+ "observed.data.table.writes.behind": "3",
+ "observed.partitions.sampled": "20",
+ "observed.partitions.inconsistent": "0"
+ }
+ }
+ ],
+ "errors": []
+}
+```
+
+`effectiveConfigs` key convention: bare `hoodie.*` keys are writer properties
as supplied;
+`effective.*` are thresholds the check derived from them; `observed.*` are
measurements from the
+table. `overallStatus` is the most severe individual status (`UNHEALTHY` >
`HEALTHY` > `SKIPPED`).
+
+All three checks report `SKIPPED` when the table has no metadata table, or
when the writer runs
+with `hoodie.metadata.enable=false`. That is the end of the conversation for
this Skill: nothing
+below applies to a table without a metadata table.
+
+## Per check
+
+### mdt-sync
+
+| | |
+|---|---|
+| Needs | `hoodie.metadata.enable` |
+| Unhealthy when | any completed data-table write is newer than the metadata
table's synced instant, **or** the latest base-file/log-file names differ
between the metadata table and storage in any sampled partition (all partitions
when 20 or fewer; otherwise the 20 newest in sorted order, where drift
originates, plus the 2 oldest) |
+| Finding "writes are missing from the metadata table" | Sync lag. If a writer
is active, a write may still be finalizing: rerun after it completes. If it
persists, look for incomplete metadata-table delta commits — `metadata timeline
show incomplete` in `hudi-cli` — and for metadata-table errors in the writer
logs. |
+| Finding "Partition '...': N base files listed by the metadata table but
absent on storage" (or the reverse, or the log-file variants) | Listing
inconsistency. The metadata table has drifted from what is on storage; readers
that list through it see the wrong files. The finding names the first few files
on each side. |
+| Remedy for lag | **Not "wait for the next write".** A later write does not
replay an instant the metadata table missed — the writer re-initializes only
when the metadata timeline is empty
(`HoodieBackedTableMetadataWriter.isBootstrapNeeded`), so the next write writes
its own delta commit and moves the synced instant past the gap. The check then
reads healthy while the missed write's files stay absent from listings. Lag
that does not correspond to an in-flight write is a possible incons [...]
+| Remedy for inconsistency | First confirm the extent: `spark-submit --class
org.apache.hudi.utilities.HoodieMetadataTableValidator <hudi-utilities-bundle>
--base-path <scheme-qualified-path> --validate-latest-file-slices
--validate-latest-base-files`. **Pass the base path with its scheme**
(`file:///data/table`, `s3://bucket/table`, `hdfs://nn/table`), not a bare
path. The validator compares the metadata-table listing against a file-system
listing by full path string: the metadata side [...]
+
+### mdt-compaction
+
+| | |
+|---|---|
+| Needs | `hoodie.metadata.enable`,
`hoodie.metadata.compact.max.delta.commits` |
+| Unhealthy when | completed metadata-table delta commits since the last
completed metadata-table compaction exceed 2x the configured trigger |
+| Also reports | `observed.lag.duration.minutes` — how long compaction has
been lagging, from the last compaction (or the first delta commit if there has
never been one) to the latest delta commit;
`observed.last.mdt.compaction.instant`;
`observed.data.table.earliest.pending.instant` |
+| Why it lags | Metadata-table compaction runs inline with the data-table
writer. To keep the compacted base file consistent with the data table, the
writer pins the compaction instant just below the earliest data-table instant
that is still pending but already applied to the metadata table. Once a
compaction at that pinned instant exists, later attempts are skipped until the
pending instant completes or is rolled back. |
+| Finding "Data-table instant ... is pending" | The first thing to check. If
the instant belongs to a writer that is gone, rolling it back unblocks
compaction. `timeline show incomplete` in `hudi-cli` lists pending data-table
instants. |
+| Finding "No data-table instant is pending" | Nothing is pinning compaction,
so look at the writer: confirm it runs metadata-table table services after its
commits, and search its logs for `Error in scheduling and executing compaction
in metadata table`. |
+| Finding "compaction(s) are scheduled on the metadata table but not complete"
| `metadata timeline show incomplete` in `hudi-cli` lists them. The writer
re-runs pending metadata-table compactions before its next write; repeated
failures point at the writer logs. |
+| Remedy | Clear whatever is pinning it — usually completing or rolling back
the pending data-table instant — and let the writer's next commit compact.
Compaction lag recovers on its own. **Do not rebuild the metadata table for
compaction lag.** Rolling back a pending instant discards only that instant's
uncommitted work. |
+
+### record-index
+
+| | |
+|---|---|
+| Applies to | tables whose `hoodie.table.metadata.partitions` lists
`record_index`; `SKIPPED` otherwise |
+| Needs | for a global index:
`hoodie.metadata.global.record.level.index.min.filegroup.count` and
`hoodie.metadata.global.record.level.index.max.filegroup.count`; for a
partitioned index (`hoodie.metadata.record.level.index.enable=true`):
`hoodie.metadata.record.level.index.min.filegroup.count` and
`hoodie.metadata.record.level.index.max.filegroup.count`; plus, in both cases,
`hoodie.metadata.record.index.max.filegroup.size` and
`hoodie.metadata.record.index.growth.factor`. Config altern [...]
+| Unhealthy when | `observed.file.group.count` is below
`effective.min.healthy.file.group.count`, which is ceil(0.5 x
`effective.ideal.file.group.count`) — per data partition for a partitioned
index, with the worst data partition reported — **or**
`observed.largest.file.group.bytes` exceeds
`effective.max.healthy.file.group.bytes`, which is 1.5 x
`hoodie.metadata.record.index.max.filegroup.size` |
+| How the ideal is derived | The record count is an **estimate**: index bytes
divided by `effective.average.record.size.bytes` (48). The check labels it as
such; treat `observed.estimated.record.count` as an order of magnitude, not a
fact. |
+| `effective.count.comparison` | Says what was compared: `whole index (global
record index)`; `per data partition (partitioned record index); worst data
partition reported`; or a `skipped: ...` string when the writer configuration
says one kind of index but the index on storage is the other. A `skipped`
comparison means the count rule did not run — say so rather than calling the
count healthy. |
+| Reports | `observed.file.group.count`, `observed.base.file.count`,
`observed.log.file.count`, `observed.total.size.bytes`,
`observed.largest.file.group.bytes`, `observed.largest.file.group.id`,
`observed.estimated.record.count`; for partitioned indexes also
`observed.data.partition.count`, `observed.worst.data.partition`,
`observed.worst.data.partition.file.group.count` |
+| Why it matters | The file-group count of the record-level index is fixed at
initialization and cannot be changed in place; it does not grow with the table.
An undersized index concentrates lookups and log files on too few file groups,
and each file group grows without bound. |
+| Remedy | Drop the index, correct the sizing configuration, and rebuild. Drop
with `metadata delete-record-index` in `hudi-cli` (or `HoodieIndexer --mode
dropindex --index-types RECORD_INDEX`); fix the file-group count and size
properties above; then rebuild with `HoodieIndexer --mode scheduleAndExecute
--index-types RECORD_INDEX`. This is **disruptive** — the index is unavailable
during the rebuild, so lookups fall back to the slower path — and **not
undoable in place**; schedule it wi [...]
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java
index fbda47f4a776..2d6625a008b3 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java
@@ -30,6 +30,8 @@ import org.apache.hudi.utilities.health.HealthCheck;
import org.apache.hudi.utilities.health.HealthCheckContext;
import org.apache.hudi.utilities.health.HealthCheckResult;
import org.apache.hudi.utilities.health.HealthStatus;
+import org.apache.hudi.utilities.health.MetadataCompactionLagCheck;
+import org.apache.hudi.utilities.health.MetadataTableSyncCheck;
import org.apache.hudi.utilities.health.SavepointArchivalBlockCheck;
import com.beust.jcommander.JCommander;
@@ -146,7 +148,9 @@ public class HoodieTableHealthChecker implements
Serializable {
new CompactionCadenceCheck(),
new CleanerCadenceCheck(),
new SavepointArchivalBlockCheck(),
- new ArchivalCadenceCheck());
+ new ArchivalCadenceCheck(),
+ new MetadataTableSyncCheck(),
+ new MetadataCompactionLagCheck());
}
/**
@@ -287,7 +291,7 @@ public class HoodieTableHealthChecker implements
Serializable {
public String basePath = null;
@Parameter(names = {"--checks"}, description = "Comma-separated checks to
run, or 'all'. Available: "
- + "compaction, cleaner, savepoint, archival")
+ + "compaction, cleaner, savepoint, archival, mdt-sync, mdt-compaction")
public String checks = "all";
@Parameter(names = {"--apply-all-defaults"}, description = "Evaluate
checks against Hudi default values for any "
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/README.md
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/README.md
index ea40c0691d49..f00d6072249d 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/README.md
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/README.md
@@ -147,6 +147,59 @@ This check does not attribute the cause. A stalled
archival is most often a save
Governing property: `hoodie.keep.max.commits`.
+### `mdt-sync`
+
+Whether the metadata table is caught up with the data table, and whether the
file listing it
+serves agrees with storage. Skipped when the table has no metadata table, or
when the writer runs
+with `hoodie.metadata.enable=false`.
+
+Two signals. **Sync lag**: every completed data-table write is mirrored by a
delta commit on the
+metadata table with the same instant time, so the latest completed
metadata-table delta commit is
+the instant the metadata table is synced to. Any completed data-table write
newer than that is one
+the metadata table does not know about; the report gives both instants and how
many writes are
+missing. **Listing consistency**: for a bounded sample of partitions, the
latest base-file and
+log-file names are listed once through the metadata table and once straight
from the file
+system, and compared. The sample is all partitions when there are 20 or fewer;
otherwise the last
+20 in sorted order — the most recently written on a date-partitioned table,
which is where drift
+from writes, cleans and rollbacks originates — plus the first 2, so drift
confined to old data is
+not invisible. The sample is bounded because a file-system listing is exactly
the cost the
+metadata table exists to avoid.
+
+Any lag or any differing partition is unhealthy. Lag alone comes with the
advice to rerun if a
+writer is active and otherwise to look for incomplete metadata-table delta
commits
+(`metadata timeline show incomplete` in `hudi-cli`); it is never a reason to
rebuild. A differing
+partition names the first few files on each side and points at the complete
comparison —
+`HoodieMetadataTableValidator` with `--validate-latest-file-slices
--validate-latest-base-files` —
+before any rebuild, which is disruptive and should be scheduled deliberately.
+
+This check lists files, so it costs more than the timeline-only checks, but it
still needs no
+Spark session.
+
+Governing property: `hoodie.metadata.enable`.
+
+### `mdt-compaction`
+
+Whether compaction on the metadata table is keeping pace with the delta
commits landing on it.
+Skipped when the table has no metadata table, or when the writer runs with
+`hoodie.metadata.enable=false`.
+
+The metadata table is a Merge-on-Read table of its own under
`.hoodie/metadata`, compacted inline
+by the data-table writer after its commits, every
`hoodie.metadata.compact.max.delta.commits`
+delta commits. The check counts completed metadata-table delta commits since
the last completed
+metadata-table compaction and applies the same 2.0 slack factor as the
`compaction` check. How
+long the lag has lasted is reported too, as the span from the last compaction
(or the first delta
+commit, if there has never been one) to the latest delta commit.
+
+The most common cause is not a failure. To keep a compacted base file
consistent with the data
+table, the writer pins the compaction instant just below the earliest
data-table instant that is
+still pending but already applied to the metadata table; once a compaction at
that pinned instant
+exists, every later attempt is skipped until the pending instant completes or
is rolled back. So
+when unhealthy, the report names the earliest pending data-table instant, if
any, as the first
+thing to check. Compaction lag recovers on its own once compaction runs; it is
never a reason to
+rebuild the metadata table.
+
+Governing properties: `hoodie.metadata.enable`,
`hoodie.metadata.compact.max.delta.commits`.
+
## Exit codes
| Code | Meaning |
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataCompactionLagCheck.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataCompactionLagCheck.java
new file mode 100644
index 000000000000..3c01e53c9315
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataCompactionLagCheck.java
@@ -0,0 +1,231 @@
+/*
+ * 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.utilities.health;
+
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.TimelineUtils;
+import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.storage.StoragePath;
+
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Checks whether compaction on the metadata table is keeping pace with the
delta commits landing
+ * on it.
+ *
+ * <p>The metadata table is a Merge-on-Read table of its own under {@code
.hoodie/metadata}, and
+ * every data-table write adds a delta commit to it. Its compaction is not a
separately scheduled
+ * service: the data-table writer runs it inline, after its own commit, when
the configured number
+ * of delta commits has accumulated. When that compaction stops running, every
metadata-table read
+ * has to merge an ever-growing chain of log files, and the data table's own
archival -- which
+ * cannot move past the last metadata-table compaction -- stalls with it.
+ *
+ * <p>The most common reason it stops is not a failure but the rule the writer
applies to pick the
+ * compaction instant. To keep a compacted base file consistent with the data
table, the writer
+ * pins the compaction instant just below the earliest data-table instant that
is still pending
+ * but whose delta commit has already been applied to the metadata table. Once
a compaction at
+ * that pinned instant exists, every later attempt lands on the same instant
and is skipped, until
+ * the pending data-table instant completes or is rolled back. A long-running
or abandoned
+ * data-table instant therefore freezes metadata-table compaction, and it is
the first thing to
+ * look at.
+ *
+ * <p>The same slack factor as the data-table compaction check is applied
before calling the table
+ * unhealthy: compaction lands one or two writes after the trigger point, and
the aim is to catch a
+ * compaction that has stopped, not scheduling jitter.
+ */
+public class MetadataCompactionLagCheck implements HealthCheck {
+
+ static final String NAME = "mdt-compaction";
+
+ /**
+ * How far past the configured trigger point delta commits may accumulate on
the metadata table
+ * before it is called unhealthy. Compaction is attempted after the write
that crosses the
+ * trigger, and one more write can land while a pending data-table instant
briefly pins it.
+ */
+ static final double TRIGGER_SLACK_FACTOR = 2.0;
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Whether compaction on the metadata table is keeping pace with its
delta commits";
+ }
+
+ @Override
+ public HealthCheckResult check(HealthCheckContext context) throws Exception {
+ Option<String> missing = context.firstMissingConfig(
+ HoodieMetadataConfig.ENABLE.key(),
HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key());
+ if (missing.isPresent()) {
+ return context.skipForMissingConfig(NAME, missing.get());
+ }
+
+ boolean metadataEnabled =
Boolean.parseBoolean(context.getProps().getString(
+ HoodieMetadataConfig.ENABLE.key(),
String.valueOf(HoodieMetadataConfig.ENABLE.defaultValue())));
+ if (!metadataEnabled) {
+ return HealthCheckResult.skipped(NAME, "Metadata table is disabled for
this writer ("
+ + HoodieMetadataConfig.ENABLE.key() + "=false), so there is no
metadata table to compact");
+ }
+
+ HoodieTableMetaClient dataMetaClient = context.getMetaClient();
+ Option<String> noMetadataTable =
MetadataTablePresence.reasonAbsent(dataMetaClient);
+ if (noMetadataTable.isPresent()) {
+ return HealthCheckResult.skipped(NAME, noMetadataTable.get());
+ }
+
+ int maxDeltaCommits = Integer.parseInt(context.getProps().getString(
+ HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key(),
+
String.valueOf(HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.defaultValue())));
+ int threshold = (int) Math.ceil(TRIGGER_SLACK_FACTOR * maxDeltaCommits);
+
+ HealthCheckResult.Builder result = HealthCheckResult.builder(NAME)
+ .config(HoodieMetadataConfig.ENABLE.key(), true)
+ .config(HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key(),
maxDeltaCommits)
+ .config("effective.unhealthy.threshold.delta.commits", threshold);
+
+ HoodieActiveTimeline metadataTimeline =
metadataTableMetaClient(dataMetaClient).getActiveTimeline();
+ Option<Pair<HoodieTimeline, HoodieInstant>> sinceLastCompaction =
+
CompactionUtils.getCompletedDeltaCommitsSinceLatestCompaction(metadataTimeline);
+ if (!sinceLastCompaction.isPresent()) {
+ return result.status(HealthStatus.HEALTHY)
+ .summary("No delta commits have landed on the metadata table yet;
nothing for compaction to fall behind on")
+ .build();
+ }
+
+ // When no compaction has completed, the instant returned is the first
delta commit, not a compaction.
+ Option<HoodieInstant> lastCompaction =
metadataTimeline.getCommitTimeline().filterCompletedInstants().lastInstant();
+ HoodieTimeline deltaCommitsSince = sinceLastCompaction.get().getLeft();
+ String lagStart = lagStartTime(lastCompaction, deltaCommitsSince);
+ Option<HoodieInstant> latestDeltaCommit = deltaCommitsSince.lastInstant();
+ int deltaCommitsSinceCompaction = deltaCommitsSince.countInstants();
+ Option<Long> lagMinutes = latestDeltaCommit.flatMap(latest ->
minutesBetween(lagStart, latest.requestedTime()));
+ int pendingMetadataCompactions =
metadataTimeline.filterPendingCompactionTimeline().countInstants();
+ // Only a pending data-table instant that is already completed in the
metadata table pins the
+ // compaction instant -- compactIfNecessary filters on exactly that
+ // (HoodieBackedTableMetadataWriter: filter(instant ->
metadataCompletedTimeline.containsInstant(...))).
+ // Taking the earliest pending instant of any action would, on a MOR table
with async table
+ // services, usually name a pending compaction or clean that is not
blocking anything, and then
+ // suggest rolling it back.
+ HoodieTimeline metadataCompletedTimeline =
metadataTimeline.filterCompletedInstants();
+ Option<HoodieInstant> earliestPendingDataInstant =
dataMetaClient.getActiveTimeline()
+ .filterInflightsAndRequested()
+ .filter(instant ->
metadataCompletedTimeline.containsInstant(instant.requestedTime()))
+ .firstInstant();
+
+ result.config("observed.mdt.delta.commits.since.compaction",
deltaCommitsSinceCompaction)
+ .config("observed.last.mdt.compaction.instant",
lastCompaction.map(HoodieInstant::requestedTime).orElse("none"))
+ .config("observed.latest.mdt.delta.commit.instant",
latestDeltaCommit.map(HoodieInstant::requestedTime).orElse("none"))
+ .config("observed.lag.duration.minutes",
lagMinutes.map(String::valueOf).orElse("unknown"))
+ .config("observed.pending.mdt.compactions", pendingMetadataCompactions)
+ .config("observed.data.table.earliest.pending.instant",
earliestPendingDataInstant.map(HoodieInstant::requestedTime).orElse("none"));
+
+ String since = lastCompaction.isPresent()
+ ? "compaction at " + lastCompaction.get().requestedTime()
+ : "the metadata table's first delta commit at " + lagStart;
+ String duration = lagMinutes.map(m -> String.format(" over roughly %d
minute(s)", m)).orElse("");
+
+ if (deltaCommitsSinceCompaction <= threshold) {
+ return result.status(HealthStatus.HEALTHY)
+ .summary(String.format("%d metadata-table delta commit(s) since %s,
within the threshold of %d",
+ deltaCommitsSinceCompaction, since, threshold))
+ .build();
+ }
+
+ result.status(HealthStatus.UNHEALTHY)
+ .summary(String.format("%d metadata-table delta commit(s) have
accumulated since %s%s, past the threshold of %d",
+ deltaCommitsSinceCompaction, since, duration, threshold))
+ .finding(String.format(
+ "The writer is configured to compact the metadata table every %d
delta commit(s); %d have accumulated%s. "
+ + "Every metadata-table read merges that many log files, and
data-table archival cannot advance past "
+ + "the last metadata-table compaction.",
+ maxDeltaCommits, deltaCommitsSinceCompaction, duration));
+
+ if (earliestPendingDataInstant.isPresent()) {
+ result.finding(String.format(
+ "Data-table instant %s (%s) is pending. Metadata-table compaction
runs inline with the data-table writer "
+ + "and pins its instant just below the earliest pending
data-table instant already applied to the "
+ + "metadata table; once a compaction at that pinned instant
exists, further attempts are skipped "
+ + "until the pending instant completes or is rolled back. Check
that instant first: if it belongs to "
+ + "a writer that is gone, rolling it back unblocks compaction
('timeline show incomplete' in hudi-cli "
+ + "lists it).",
+ earliestPendingDataInstant.get().requestedTime(),
earliestPendingDataInstant.get().getAction()));
+ } else {
+ result.finding("No pending data-table instant has been applied to the
metadata table, so nothing is pinning "
+ + "the compaction instant. Confirm the "
+ + "writer runs metadata-table table services after its commits, and
look in the writer logs for "
+ + "'Error in scheduling and executing compaction in metadata
table'.");
+ }
+
+ if (pendingMetadataCompactions > 0) {
+ result.finding(String.format(
+ "%d compaction(s) are scheduled on the metadata table but not
complete ('metadata timeline show incomplete' "
+ + "in hudi-cli lists them). The writer re-runs pending
metadata-table compactions before its next "
+ + "write; repeated failures there point at the writer logs.",
+ pendingMetadataCompactions));
+ }
+
+ return result.finding("Do not delete or rebuild the metadata table for
this condition; it recovers on its own "
+ + "once compaction runs.").build();
+ }
+
+ private HoodieTableMetaClient metadataTableMetaClient(HoodieTableMetaClient
dataMetaClient) {
+ StoragePath metadataTablePath =
HoodieTableMetadata.getMetadataTableBasePath(dataMetaClient.getBasePath());
+ return HoodieTableMetaClient.builder()
+ .setConf(dataMetaClient.getStorageConf().newInstance())
+ .setBasePath(metadataTablePath)
+ .setLoadActiveTimelineOnLoad(true)
+ .build();
+ }
+
+ /**
+ * Where the lag is measured from: the last compaction, or, when there has
never been one, the
+ * first delta commit whose time is a real timestamp. The metadata table's
bootstrap delta commit
+ * carries a zero placeholder instant that would otherwise make every
never-compacted table look
+ * infinitely behind.
+ */
+ private static String lagStartTime(Option<HoodieInstant> lastCompaction,
HoodieTimeline deltaCommitsSince) {
+ if (lastCompaction.isPresent()) {
+ return lastCompaction.get().requestedTime();
+ }
+ return deltaCommitsSince.getInstantsAsStream()
+ .map(HoodieInstant::requestedTime)
+ .filter(time ->
TimelineUtils.parseDateFromInstantTimeSafely(time).isPresent())
+ .findFirst()
+ .orElse(HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP);
+ }
+
+ private static Option<Long> minutesBetween(String fromInstant, String
toInstant) {
+ Option<Date> from =
TimelineUtils.parseDateFromInstantTimeSafely(fromInstant);
+ Option<Date> to = TimelineUtils.parseDateFromInstantTimeSafely(toInstant);
+ if (!from.isPresent() || !to.isPresent()) {
+ return Option.empty();
+ }
+ return Option.of(TimeUnit.MILLISECONDS.toMinutes(to.get().getTime() -
from.get().getTime()));
+ }
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataTablePresence.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataTablePresence.java
new file mode 100644
index 000000000000..b85555c9b5c1
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataTablePresence.java
@@ -0,0 +1,51 @@
+/*
+ * 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.utilities.health;
+
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+
+/**
+ * Whether a metadata table exists for the data table.
+ *
+ * <p>Shared by the metadata-table checks so they skip for the same reasons in
the same words, and
+ * so neither depends on the other's internals.
+ */
+final class MetadataTablePresence {
+
+ private MetadataTablePresence() {
+ }
+
+ /** Empty when the metadata table is present; otherwise the reason to skip,
for the operator. */
+ static Option<String> reasonAbsent(HoodieTableMetaClient metaClient) throws
IOException {
+ if (!metaClient.getTableConfig().isMetadataTableAvailable()) {
+ return Option.of("Table has no metadata table (its FILES partition has
not been initialized)");
+ }
+ StoragePath metadataTablePath =
HoodieTableMetadata.getMetadataTableBasePath(metaClient.getBasePath());
+ if (!metaClient.getStorage().exists(metadataTablePath)) {
+ return Option.of("Metadata table is registered in hoodie.properties but
" + metadataTablePath
+ + " does not exist on storage; the next write will re-initialize
it");
+ }
+ return Option.empty();
+ }
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataTableSyncCheck.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataTableSyncCheck.java
new file mode 100644
index 000000000000..f20606186470
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/MetadataTableSyncCheck.java
@@ -0,0 +1,430 @@
+/*
+ * 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.utilities.health;
+
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.HoodieLocalEngineContext;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.TimelineUtils;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.metadata.FileSystemBackedTableMetadata;
+import org.apache.hudi.metadata.HoodieBackedTableMetadata;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+
+/**
+ * Checks whether the metadata table is caught up with the data table, and
whether the file listing
+ * it serves agrees with what is actually on storage.
+ *
+ * <p>Two signals. The first is sync lag: every completed data-table write is
mirrored by a delta
+ * commit on the metadata table with the same instant time, so the latest
completed metadata-table
+ * delta commit says how far the metadata table has been brought up to date. A
data-table write
+ * newer than that is one the metadata table does not know about, and readers
relying on the
+ * metadata table for listings will not see its files.
+ *
+ * <p>The second is listing consistency: for a bounded sample of partitions,
the latest base files
+ * and log files are listed twice -- once through the metadata table and once
straight from the
+ * file system -- and compared. A disagreement means the metadata table has
drifted from storage,
+ * which is the one condition that justifies rebuilding it. The sample is
bounded because listing a
+ * partition from the file system is precisely the expensive operation the
metadata table exists to
+ * avoid; a daily health check must not turn into a full table scan. It is
weighted towards the
+ * most recently written partitions, where drift actually originates, with a
couple of the oldest
+ * added for coverage. For a complete comparison the finding points at
+ * {@code HoodieMetadataTableValidator}.
+ *
+ * <p>Engine-agnostic: both listing backends and the file-system views are
built on
+ * {@link HoodieLocalEngineContext}, so no Spark session is needed.
+ */
+public class MetadataTableSyncCheck implements HealthCheck {
+
+ static final String NAME = "mdt-sync";
+
+ /**
+ * Upper bound on partitions whose listings are compared. Each sampled
partition costs one
+ * file-system listing, which is the operation the metadata table exists to
avoid, so the sample
+ * is kept small enough to be cheap on a schedule while still catching drift
that affects the
+ * table broadly rather than a single partition.
+ *
+ * <p>The sample is taken from the <em>end</em> of the sorted partition
list. Drift comes from
+ * recent writes, cleans and rollbacks, so the partitions most likely to
disagree are the ones
+ * most recently written -- and on the common date-partitioned table,
lexical order is time
+ * order, so the last partitions are the newest. Sampling the first ones
would compare the
+ * partitions least likely to have changed.
+ */
+ static final int MAX_PARTITIONS_TO_COMPARE = 20;
+
+ /**
+ * How many of the oldest partitions are added to the sample when the table
has more partitions
+ * than {@link #MAX_PARTITIONS_TO_COMPARE}, so that drift confined to old
data (a bad clean, a
+ * partition dropped by hand) is not invisible to a check that otherwise
looks only at new data.
+ */
+ static final int OLDEST_PARTITIONS_TO_INCLUDE = 2;
+
+ /**
+ * How many differing file names to spell out per partition before
truncating. Enough to identify
+ * the pattern (one file, one file group, everything) without flooding the
report.
+ */
+ static final int MAX_DIFFERING_FILES_TO_NAME = 5;
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Whether the metadata table is caught up with the data table and
its file listing matches storage";
+ }
+
+ @Override
+ public HealthCheckResult check(HealthCheckContext context) throws Exception {
+ Option<String> missing =
context.firstMissingConfig(HoodieMetadataConfig.ENABLE.key());
+ if (missing.isPresent()) {
+ return context.skipForMissingConfig(NAME, missing.get());
+ }
+
+ boolean metadataEnabled =
Boolean.parseBoolean(context.getProps().getString(
+ HoodieMetadataConfig.ENABLE.key(),
String.valueOf(HoodieMetadataConfig.ENABLE.defaultValue())));
+
+ HoodieTableMetaClient metaClient = context.getMetaClient();
+ Option<String> noMetadataTable =
MetadataTablePresence.reasonAbsent(metaClient);
+
+ if (!metadataEnabled) {
+ // Order matters: a writer with the metadata table disabled is only
uninteresting when no
+ // metadata table exists. When one is still on storage -- which is what
happens whenever
+ // hoodie.metadata.auto.delete.partitions is false -- nothing maintains
it while readers that
+ // enable the metadata table keep serving listings from it. Skipping
here would stay silent on
+ // the one configuration in this check's remit that actively returns
wrong results.
+ if (noMetadataTable.isPresent()) {
+ return HealthCheckResult.skipped(NAME, "Metadata table is disabled for
this writer ("
+ + HoodieMetadataConfig.ENABLE.key() + "=false) and none exists on
storage, so there is "
+ + "nothing to compare against");
+ }
+ return HealthCheckResult.builder(NAME)
+ .config(HoodieMetadataConfig.ENABLE.key(), false)
+ .status(HealthStatus.UNHEALTHY)
+ .summary("Metadata table exists on storage but the writer has it
disabled, so nothing is "
+ + "maintaining it")
+ .finding("The supplied writer properties set " +
HoodieMetadataConfig.ENABLE.key()
+ + "=false, yet a metadata table is present. It is not being
updated by this writer, so "
+ + "it falls further behind with every commit.")
+ .finding("Readers are not bound by the writer's setting. Any reader
that enables the "
+ + "metadata table will serve file listings from this stale copy
and silently miss data "
+ + "written since the writer stopped maintaining it.")
+ .finding("Decide which you want. To stop using it, delete it
explicitly ("
+ + "`metadata delete` in hudi-cli) rather than leaving it in
place -- note that "
+ + HoodieMetadataConfig.AUTO_DELETE_PARTITIONS.key() + " defaults
to true, so a write "
+ + "with the metadata table disabled normally removes it, and its
being here means that "
+ + "was turned off. To resume using it, re-enable it on the
writer and let it rebuild.")
+ .build();
+ }
+
+ if (noMetadataTable.isPresent()) {
+ return HealthCheckResult.skipped(NAME, noMetadataTable.get());
+ }
+
+ HealthCheckResult.Builder result = HealthCheckResult.builder(NAME)
+ .config(HoodieMetadataConfig.ENABLE.key(), true)
+ .config("effective.partitions.sampled.max", MAX_PARTITIONS_TO_COMPARE);
+
+ HoodieEngineContext engineContext = new
HoodieLocalEngineContext(metaClient.getStorageConf());
+ String basePath = metaClient.getBasePath().toString();
+ HoodieMetadataConfig metadataConfig =
HoodieMetadataConfig.newBuilder().enable(true).build();
+
+ try (HoodieTableMetadata metadataBacked =
+ new HoodieBackedTableMetadata(engineContext,
metaClient.getStorage(), metadataConfig, basePath);
+ HoodieTableMetadata fileSystemBacked =
+ new FileSystemBackedTableMetadata(engineContext,
metaClient.getTableConfig(), metaClient.getStorage(), basePath)) {
+
+ SyncLag lag = measureSyncLag(metaClient, metadataBacked);
+ lag.report(result);
+
+ ListingComparison listing = compareSampledPartitions(metaClient,
metadataBacked, fileSystemBacked);
+ listing.report(result);
+
+ return verdict(result, metaClient, lag, listing);
+ }
+ }
+
+ private HealthCheckResult verdict(HealthCheckResult.Builder result,
HoodieTableMetaClient metaClient,
+ SyncLag lag, ListingComparison listing) {
+ boolean behind = lag.commitsBehind > 0;
+ boolean inconsistent = !listing.differences.isEmpty();
+
+ if (!behind && !inconsistent) {
+ return result.status(HealthStatus.HEALTHY)
+ .summary(String.format("Metadata table is synced to %s and its
listing matches storage in all %d sampled partition(s)",
+ lag.syncedInstant.orElse("(no data-table writes yet)"),
listing.partitionsSampled))
+ .build();
+ }
+
+ if (behind) {
+ result.finding(String.format(
+ "The metadata table is synced to %s but the data table's latest
completed write is %s; %d completed "
+ + "data-table write(s) are missing from the metadata table.
Readers that list through the "
+ + "metadata table will not see files from those writes.",
+ lag.syncedInstant.orElse("nothing"),
lag.latestDataInstant.orElse("nothing"), lag.commitsBehind));
+ result.finding("If a writer is running right now, this can be a write
that is still being finalized: "
+ + "rerun once it completes. Look for failed or incomplete
metadata-table delta commits "
+ + "('metadata timeline show incomplete' in hudi-cli) and for
metadata-table errors in the "
+ + "writer logs.");
+ result.finding("Do not wait for the lag to clear on its own. A later
write does not replay an "
+ + "instant the metadata table missed: the writer only re-initializes
when the metadata "
+ + "timeline is empty
(HoodieBackedTableMetadataWriter.isBootstrapNeeded), so the next write "
+ + "simply writes its own delta commit and moves the synced instant
past the gap. This check "
+ + "will then read as healthy while the missed write's files stay
absent from metadata-table "
+ + "listings.");
+ result.finding("Treat lag that does not correspond to an in-flight write
as a possible "
+ + "inconsistency rather than a delay. Confirm with
HoodieMetadataTableValidator "
+ + "(spark-submit, hudi-utilities) using
--validate-latest-file-slices "
+ + "--validate-latest-base-files, which compares listings rather than
instant times, and act "
+ + "on what it reports. Pass --base-path with its scheme (" +
schemeQualifiedBasePath(metaClient)
+ + "): the validator compares full path strings, and a bare path
makes every file look "
+ + "different from the file-system listing, reporting drift on a
healthy table.");
+ }
+
+ for (PartitionDifference difference : listing.differences) {
+ result.finding(difference.describe());
+ }
+ if (inconsistent) {
+ result.finding("The metadata table's listing has drifted from storage.
Confirm the extent with the full "
+ + "comparison before acting: run HoodieMetadataTableValidator
(spark-submit, hudi-utilities) with "
+ + "--base-path " + schemeQualifiedBasePath(metaClient) + "
--validate-latest-file-slices "
+ + "--validate-latest-base-files. The scheme matters: the validator
compares full path strings, "
+ + "and a bare path makes every file look different from the
file-system listing, reporting "
+ + "drift even on a healthy table.");
+ result.finding("If it confirms the drift, rebuild the metadata table
with `metadata delete` then "
+ + "`metadata create` in hudi-cli, with writers stopped. Do not
rebuild by disabling the "
+ + "metadata table for one write: that only deletes it when "
+ + HoodieMetadataConfig.AUTO_DELETE_PARTITIONS.key() + " is true (the
default), and when it is "
+ + "false the write rebuilds nothing and merely adds another instant
the metadata table has "
+ + "missed.");
+ result.finding("Either way the rebuild drops every index the metadata
table holds -- record "
+ + "index, column stats, bloom filters, secondary indexes -- not only
the FILES partition. "
+ + "What comes back is whatever the writer properties enable, so
confirm those settings "
+ + "before recreating, and expect to rebuild the other indexes. It is
a full listing and "
+ + "re-indexing of the table, so schedule it deliberately.");
+ }
+
+ String summary = inconsistent
+ ? String.format("Metadata table listing differs from storage in %d of
%d sampled partition(s)%s",
+ listing.differences.size(), listing.partitionsSampled,
+ behind ? ", and it is " + lag.commitsBehind + " write(s) behind" :
"")
+ : String.format("Metadata table is %d completed data-table write(s)
behind (synced to %s, latest write %s)",
+ lag.commitsBehind, lag.syncedInstant.orElse("nothing"),
lag.latestDataInstant.orElse("nothing"));
+ return result.status(HealthStatus.UNHEALTHY).summary(summary).build();
+ }
+
+ /**
+ * The table's base path with a scheme, for the validator command this check
recommends.
+ *
+ * <p>{@code HoodieMetadataTableValidator} compares the metadata-table
listing against a
+ * file-system listing by full path string. The metadata side reconstructs
paths from the base
+ * path it was given ({@code HoodieMetadataPayload.getFileList}), while the
file-system side is
+ * returned scheme-qualified by the filesystem itself. A bare {@code
--base-path} therefore makes
+ * every file compare unequal and the validator reports drift on a table
that is fine. The
+ * validator's own tests pass {@code "file:" + basePath} for this reason.
+ */
+ private static String schemeQualifiedBasePath(HoodieTableMetaClient
metaClient) {
+ StoragePath basePath = metaClient.getBasePath();
+ URI uri = basePath.toUri();
+ if (uri.getScheme() != null) {
+ return basePath.toString();
+ }
+ return metaClient.getStorage().getScheme() + ":" + basePath;
+ }
+
+ private SyncLag measureSyncLag(HoodieTableMetaClient metaClient,
HoodieTableMetadata metadataBacked) {
+ HoodieTimeline completedDataWrites =
metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ Option<String> latestDataInstant =
completedDataWrites.lastInstant().map(HoodieInstant::requestedTime);
+ // The FILES-partition bootstrap delta commit is stamped from
SOLO_COMMIT_TIMESTAMP rather than
+ // a real clock time. A metadata table created with no write after it --
`metadata create` in
+ // hudi-cli, or a write that triggered initialization and then failed --
reports that
+ // placeholder as its synced instant. Every real instant sorts after it,
so counting from it
+ // would call every completed commit behind, when the bootstrap listing
already covers them.
+ // MetadataCompactionLagCheck.lagStartTime skips the same placeholder.
+ Option<String> reportedSyncedInstant =
metadataBacked.getSyncedInstantTime();
+ Option<String> syncedInstant = reportedSyncedInstant.isPresent()
+ &&
!HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP.equals(reportedSyncedInstant.get())
+ &&
TimelineUtils.parseDateFromInstantTimeSafely(reportedSyncedInstant.get()).isPresent()
+ ? reportedSyncedInstant
+ : Option.empty();
+
+ int commitsBehind;
+ if (!latestDataInstant.isPresent()) {
+ commitsBehind = 0;
+ } else if (!syncedInstant.isPresent()) {
+ // Either the metadata table reports no synced instant, or it reports
the bootstrap
+ // placeholder. In both cases the initial listing covers what exists, so
there is no measured
+ // lag to report; a real inconsistency would surface through the listing
comparison instead.
+ commitsBehind = 0;
+ } else {
+ commitsBehind =
completedDataWrites.findInstantsAfter(syncedInstant.get()).countInstants();
+ }
+ return new SyncLag(latestDataInstant, syncedInstant, commitsBehind);
+ }
+
+ /**
+ * Compares the latest base-file and log-file names per sampled partition
between the two listing
+ * backends. Both views are opened over the same completed-commits timeline
so that a write landing
+ * mid-check cannot show up as a difference.
+ */
+ private ListingComparison compareSampledPartitions(HoodieTableMetaClient
metaClient,
+ HoodieTableMetadata
metadataBacked,
+ HoodieTableMetadata
fileSystemBacked) throws IOException {
+ List<String> sampled =
samplePartitions(metadataBacked.getAllPartitionPaths());
+
+ HoodieTimeline visibleTimeline =
metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
+ List<PartitionDifference> differences = new ArrayList<>();
+ try (HoodieTableFileSystemView metadataView = new
HoodieTableFileSystemView(metadataBacked, metaClient, visibleTimeline);
+ HoodieTableFileSystemView fileSystemView = new
HoodieTableFileSystemView(fileSystemBacked, metaClient, visibleTimeline)) {
+ for (String partition : sampled) {
+ PartitionDifference difference = new PartitionDifference(partition,
+ latestBaseFileNames(metadataView, partition),
latestBaseFileNames(fileSystemView, partition),
+ latestLogFileNames(metadataView, partition),
latestLogFileNames(fileSystemView, partition));
+ if (difference.isPresent()) {
+ differences.add(difference);
+ }
+ }
+ }
+ return new ListingComparison(sampled.size(), differences);
+ }
+
+ /**
+ * The partitions to compare: all of them when there are few, otherwise the
newest
+ * {@link #MAX_PARTITIONS_TO_COMPARE} in sorted order plus the {@link
#OLDEST_PARTITIONS_TO_INCLUDE}
+ * oldest. Sorted, so the result is stable from one run to the next.
+ */
+ static List<String> samplePartitions(List<String> allPartitions) {
+ List<String> sorted = new ArrayList<>(allPartitions);
+ Collections.sort(sorted);
+ if (sorted.size() <= MAX_PARTITIONS_TO_COMPARE) {
+ return sorted;
+ }
+ List<String> sampled = new ArrayList<>(sorted.subList(0,
OLDEST_PARTITIONS_TO_INCLUDE));
+ sampled.addAll(sorted.subList(sorted.size() - MAX_PARTITIONS_TO_COMPARE,
sorted.size()));
+ return sampled;
+ }
+
+ private static Set<String> latestBaseFileNames(HoodieTableFileSystemView
view, String partition) {
+ return
view.getLatestBaseFiles(partition).map(HoodieBaseFile::getFileName).collect(Collectors.toCollection(TreeSet::new));
+ }
+
+ private static Set<String> latestLogFileNames(HoodieTableFileSystemView
view, String partition) {
+ return view.getLatestFileSlices(partition).flatMap(FileSlice::getLogFiles)
+
.map(HoodieLogFile::getFileName).collect(Collectors.toCollection(TreeSet::new));
+ }
+
+ private static final class SyncLag {
+ private final Option<String> latestDataInstant;
+ private final Option<String> syncedInstant;
+ private final int commitsBehind;
+
+ private SyncLag(Option<String> latestDataInstant, Option<String>
syncedInstant, int commitsBehind) {
+ this.latestDataInstant = latestDataInstant;
+ this.syncedInstant = syncedInstant;
+ this.commitsBehind = commitsBehind;
+ }
+
+ private void report(HealthCheckResult.Builder result) {
+ result.config("observed.data.table.latest.completed.instant",
latestDataInstant.orElse("none"))
+ .config("observed.mdt.synced.instant", syncedInstant.orElse("none"))
+ .config("observed.data.table.writes.behind", commitsBehind);
+ }
+ }
+
+ private static final class ListingComparison {
+ private final int partitionsSampled;
+ private final List<PartitionDifference> differences;
+
+ private ListingComparison(int partitionsSampled, List<PartitionDifference>
differences) {
+ this.partitionsSampled = partitionsSampled;
+ this.differences = differences;
+ }
+
+ private void report(HealthCheckResult.Builder result) {
+ result.config("observed.partitions.sampled", partitionsSampled)
+ .config("observed.partitions.inconsistent", differences.size());
+ }
+ }
+
+ /** What one partition's two listings disagree on, if anything. */
+ private static final class PartitionDifference {
+ private final String partition;
+ private final Set<String> baseFilesOnlyInMetadata;
+ private final Set<String> baseFilesOnlyOnStorage;
+ private final Set<String> logFilesOnlyInMetadata;
+ private final Set<String> logFilesOnlyOnStorage;
+
+ private PartitionDifference(String partition, Set<String>
metadataBaseFiles, Set<String> storageBaseFiles,
+ Set<String> metadataLogFiles, Set<String>
storageLogFiles) {
+ this.partition = partition;
+ this.baseFilesOnlyInMetadata = difference(metadataBaseFiles,
storageBaseFiles);
+ this.baseFilesOnlyOnStorage = difference(storageBaseFiles,
metadataBaseFiles);
+ this.logFilesOnlyInMetadata = difference(metadataLogFiles,
storageLogFiles);
+ this.logFilesOnlyOnStorage = difference(storageLogFiles,
metadataLogFiles);
+ }
+
+ private static Set<String> difference(Set<String> left, Set<String> right)
{
+ Set<String> onlyInLeft = new TreeSet<>(left);
+ onlyInLeft.removeAll(right);
+ return onlyInLeft;
+ }
+
+ private boolean isPresent() {
+ return !baseFilesOnlyInMetadata.isEmpty() ||
!baseFilesOnlyOnStorage.isEmpty()
+ || !logFilesOnlyInMetadata.isEmpty() ||
!logFilesOnlyOnStorage.isEmpty();
+ }
+
+ private String describe() {
+ StringBuilder sb = new StringBuilder(String.format("Partition '%s': ",
partition.isEmpty() ? "(non-partitioned)" : partition));
+ appendIfAny(sb, "base files listed by the metadata table but absent on
storage", baseFilesOnlyInMetadata);
+ appendIfAny(sb, "base files on storage but missing from the metadata
table", baseFilesOnlyOnStorage);
+ appendIfAny(sb, "log files listed by the metadata table but absent on
storage", logFilesOnlyInMetadata);
+ appendIfAny(sb, "log files on storage but missing from the metadata
table", logFilesOnlyOnStorage);
+ return sb.toString();
+ }
+
+ private static void appendIfAny(StringBuilder sb, String label,
Set<String> files) {
+ if (files.isEmpty()) {
+ return;
+ }
+ List<String> named =
files.stream().limit(MAX_DIFFERING_FILES_TO_NAME).collect(Collectors.toList());
+ String more = files.size() > named.size() ? String.format(" (+%d more)",
files.size() - named.size()) : "";
+ sb.append(String.format("%d %s %s%s. ", files.size(), label,
String.join(", ", named), more));
+ }
+ }
+
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTableHealthChecker.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTableHealthChecker.java
index 2f194bc4e2c3..9bd07ef32fe8 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTableHealthChecker.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTableHealthChecker.java
@@ -215,7 +215,7 @@ public class TestHoodieTableHealthChecker {
JsonNode root = MAPPER.readTree(result.output);
assertEquals("UNHEALTHY", root.get("overallStatus").asText());
assertEquals(HoodieTableType.MERGE_ON_READ.name(),
root.get("tableType").asText());
- assertEquals(4, root.get("checks").size());
+ assertEquals(6, root.get("checks").size());
JsonNode compaction = root.get("checks").get(0);
assertEquals("compaction", compaction.get("name").asText());
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestMetadataCompactionLagCheck.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestMetadataCompactionLagCheck.java
new file mode 100644
index 000000000000..4e6bd40fdd82
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestMetadataCompactionLagCheck.java
@@ -0,0 +1,294 @@
+/*
+ * 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.utilities.health;
+
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.TimelineUtils;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.MetadataPartitionType;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.Date;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link MetadataCompactionLagCheck}.
+ *
+ * <p>The check reads only the metadata table's timeline, so the fixture
initializes an empty
+ * metadata table under {@code .hoodie/metadata}, registers its FILES
partition on the data table,
+ * and drives its timeline with {@link HoodieTestTable}. Completion times are
always explicit: delta
+ * commits are counted by completion time relative to the last compaction, so
a fixture that leaves
+ * them implicit can silently count nothing.
+ */
+public class TestMetadataCompactionLagCheck {
+
+ private static final String COMPACTION_INSTANT = "20260101100000000";
+
+ @TempDir
+ Path tempDir;
+
+ private String basePath;
+ private final MetadataCompactionLagCheck check = new
MetadataCompactionLagCheck();
+
+ @BeforeEach
+ public void setUp() {
+ basePath = tempDir.resolve("table").toString();
+ }
+
+ private TypedProperties propsWithTrigger(int maxDeltaCommits) {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(HoodieMetadataConfig.ENABLE.key(), "true");
+ props.setProperty(HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key(),
String.valueOf(maxDeltaCommits));
+ return props;
+ }
+
+ private HealthCheckResult runCheck(HoodieTableMetaClient metaClient,
TypedProperties props) throws Exception {
+ return check.check(new
HealthCheckContext(HoodieTableMetaClient.reload(metaClient), props, false));
+ }
+
+ private HoodieTableMetaClient tableWithoutMetadataTable() throws Exception {
+ return HoodieTestUtils.init(basePath, HoodieTableType.MERGE_ON_READ);
+ }
+
+ /** A data table with an initialized, empty metadata table registered in its
hoodie.properties. */
+ private HoodieTableMetaClient tableWithMetadataTable() throws Exception {
+ HoodieTableMetaClient dataMetaClient = HoodieTestUtils.init(basePath,
HoodieTableType.MERGE_ON_READ);
+ Properties registration = new Properties();
+
registration.setProperty(HoodieTableConfig.TABLE_METADATA_PARTITIONS.key(),
MetadataPartitionType.FILES.getPartitionPath());
+ HoodieTableConfig.update(dataMetaClient.getStorage(),
dataMetaClient.getMetaPath(), registration);
+
HoodieTestUtils.init(HoodieTableMetadata.getMetadataTableBasePath(basePath),
HoodieTableType.MERGE_ON_READ);
+ return dataMetaClient;
+ }
+
+ private HoodieTestTable metadataTableOf(String dataBasePath) throws
Exception {
+ HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder()
+ .setConf(HoodieTestUtils.getDefaultStorageConf())
+
.setBasePath(HoodieTableMetadata.getMetadataTableBasePath(dataBasePath))
+ .build();
+ return HoodieTestTable.of(metadataMetaClient);
+ }
+
+ /** The instant {@code minutes} minutes after {@link #COMPACTION_INSTANT},
formatted as an instant time. */
+ private static String minutesAfterCompaction(int minutes) throws Exception {
+ Date compactedAt =
TimelineUtils.parseDateFromInstantTime(COMPACTION_INSTANT);
+ return TimelineUtils.formatDate(new Date(compactedAt.getTime() +
TimeUnit.MINUTES.toMillis(minutes)));
+ }
+
+ private void addCompactionThenDeltaCommits(HoodieTestTable metadataTable,
int deltaCommits, int minutesApart) throws Exception {
+ metadataTable.addCompaction(COMPACTION_INSTANT,
Option.of(COMPACTION_INSTANT), new HoodieCommitMetadata());
+ addDeltaCommits(metadataTable, deltaCommits, minutesApart);
+ }
+
+ private void addDeltaCommits(HoodieTestTable metadataTable, int
deltaCommits, int minutesApart) throws Exception {
+ for (int i = 1; i <= deltaCommits; i++) {
+ String instant = minutesAfterCompaction(i * minutesApart);
+ metadataTable.addDeltaCommit(instant, Option.of(instant), new
HoodieCommitMetadata());
+ }
+ }
+
+ @Test
+ public void tableWithoutMetadataTableIsSkipped() throws Exception {
+ // given a table that has never had a metadata table
+ HoodieTableMetaClient metaClient = tableWithoutMetadataTable();
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(10));
+
+ // then there is no metadata table to compact, and the check says so
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains("no metadata table"),
result.getSummary());
+ }
+
+ @Test
+ public void metadataTableDisabledForWriterIsSkipped() throws Exception {
+ // given a table with a metadata table
+ HoodieTableMetaClient metaClient = tableWithMetadataTable();
+ TypedProperties props = propsWithTrigger(10);
+ props.setProperty(HoodieMetadataConfig.ENABLE.key(), "false");
+
+ // when the writer is configured with the metadata table off
+ HealthCheckResult result = runCheck(metaClient, props);
+
+ // then the check does not apply
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+
assertTrue(result.getSummary().contains(HoodieMetadataConfig.ENABLE.key()),
result.getSummary());
+ }
+
+ @Test
+ public void missingTriggerPropertySkipsNamingIt() throws Exception {
+ // given a lagging metadata table
+ HoodieTableMetaClient metaClient = tableWithMetadataTable();
+ addCompactionThenDeltaCommits(metadataTableOf(basePath), 30, 1);
+ TypedProperties onlyEnable = new TypedProperties();
+ onlyEnable.setProperty(HoodieMetadataConfig.ENABLE.key(), "true");
+
+ // when the check runs without the compaction trigger configured
+ HealthCheckResult result = runCheck(metaClient, onlyEnable);
+
+ // then it declines to judge rather than guessing a trigger the table may
not be written with
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+
assertTrue(result.getSummary().contains(HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key()),
result.getSummary());
+ }
+
+ @Test
+ public void metadataTableWithNoDeltaCommitsIsHealthy() throws Exception {
+ // given a freshly initialized metadata table with an empty timeline
+ HoodieTableMetaClient metaClient = tableWithMetadataTable();
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(10));
+
+ // then nothing has accumulated
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ }
+
+ @Test
+ public void deltaCommitsWithinSlackAreHealthy() throws Exception {
+ // given 8 delta commits since compaction against a trigger of 5 -- past
the trigger, inside the 2x slack
+ HoodieTableMetaClient metaClient = tableWithMetadataTable();
+ addCompactionThenDeltaCommits(metadataTableOf(basePath), 8, 1);
+
+ // when the check runs (threshold is 10)
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(5));
+
+ // then the gap between trigger and execution is not an alert
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ assertEquals("8",
result.getEffectiveConfigs().get("observed.mdt.delta.commits.since.compaction"));
+ assertEquals("10",
result.getEffectiveConfigs().get("effective.unhealthy.threshold.delta.commits"));
+ }
+
+ @Test
+ public void deltaCommitsPastSlackAreUnhealthyWithDurationEchoed() throws
Exception {
+ // given a trigger of 2 and 6 delta commits landing every 10 minutes after
the last compaction
+ HoodieTableMetaClient metaClient = tableWithMetadataTable();
+ addCompactionThenDeltaCommits(metadataTableOf(basePath), 6, 10);
+
+ // when the check runs (threshold is 4)
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(2));
+
+ // then compaction is reported as lagging, with how long it has been
lagging
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertEquals("6",
result.getEffectiveConfigs().get("observed.mdt.delta.commits.since.compaction"));
+ assertEquals(COMPACTION_INSTANT,
result.getEffectiveConfigs().get("observed.last.mdt.compaction.instant"));
+ assertEquals("60",
result.getEffectiveConfigs().get("observed.lag.duration.minutes"));
+ assertTrue(result.getSummary().contains("60 minute"), result.getSummary());
+ }
+
+ @Test
+ public void pendingDataTableInstantIsNamedAsTheFirstThingToCheck() throws
Exception {
+ // given a lagging metadata table and a data-table delta commit stuck
inflight
+ HoodieTableMetaClient metaClient = tableWithMetadataTable();
+ addCompactionThenDeltaCommits(metadataTableOf(basePath), 6, 10);
+ String pendingInstant = minutesAfterCompaction(30);
+ HoodieTestTable.of(metaClient).addInflightDeltaCommit(pendingInstant);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(2));
+
+ // then the pending instant is named, since it is what pins the compaction
instant
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertEquals(pendingInstant,
result.getEffectiveConfigs().get("observed.data.table.earliest.pending.instant"));
+ assertTrue(result.getFindings().stream().anyMatch(f ->
f.contains(pendingInstant) && f.contains("pending")),
+ "expected the pending data-table instant to be named, got: " +
result.getFindings());
+ }
+
+ @Test
+ public void unhealthyResultNeverRecommendsRebuildingTheMetadataTable()
throws Exception {
+ // given a lagging metadata table
+ HoodieTableMetaClient metaClient = tableWithMetadataTable();
+ addCompactionThenDeltaCommits(metadataTableOf(basePath), 6, 10);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(2));
+
+ // then the operator is told compaction lag recovers on its own
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("Do not
delete or rebuild")),
+ "expected a warning against rebuilding, got: " + result.getFindings());
+ }
+
+ @Test
+ public void
metadataTableThatHasNeverCompactedCountsFromItsFirstDeltaCommit() throws
Exception {
+ // given 6 delta commits and no compaction at all
+ HoodieTableMetaClient metaClient = tableWithMetadataTable();
+ addDeltaCommits(metadataTableOf(basePath), 6, 10);
+
+ // when the check runs with a trigger of 2
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(2));
+
+ // then every delta commit counts, and the report does not pretend a
compaction happened
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertEquals("6",
result.getEffectiveConfigs().get("observed.mdt.delta.commits.since.compaction"));
+ assertEquals("none",
result.getEffectiveConfigs().get("observed.last.mdt.compaction.instant"));
+ assertFalse(result.getSummary().contains("compaction at"),
result.getSummary());
+ }
+
+ @Test
+ public void bootstrapPlaceholderInstantDoesNotDistortTheLagDuration() throws
Exception {
+ // given a never-compacted metadata table whose first delta commit is the
zero bootstrap placeholder
+ HoodieTableMetaClient metaClient = tableWithMetadataTable();
+ HoodieTestTable metadataTable = metadataTableOf(basePath);
+ String bootstrapInstant = HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP +
"000";
+ metadataTable.addDeltaCommit(bootstrapInstant,
Option.of(bootstrapInstant), new HoodieCommitMetadata());
+ addDeltaCommits(metadataTable, 6, 10);
+
+ // when the check runs with a trigger of 2
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(2));
+
+ // then the duration is measured from the first real delta commit, not
from the placeholder
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertEquals("7",
result.getEffectiveConfigs().get("observed.mdt.delta.commits.since.compaction"));
+ assertEquals("50",
result.getEffectiveConfigs().get("observed.lag.duration.minutes"));
+ assertFalse(result.getSummary().contains(bootstrapInstant),
result.getSummary());
+ }
+
+ @Test
+ public void effectiveConfigIsEchoedSoAVerdictCanBeAudited() throws Exception
{
+ // given a healthy metadata table
+ HoodieTableMetaClient metaClient = tableWithMetadataTable();
+ addCompactionThenDeltaCommits(metadataTableOf(basePath), 3, 1);
+
+ // when the check runs with a known trigger
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(10));
+
+ // then the supplied value, the derived threshold, and the measurements
are all reported
+ assertEquals("true",
result.getEffectiveConfigs().get(HoodieMetadataConfig.ENABLE.key()));
+ assertEquals("10",
result.getEffectiveConfigs().get(HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key()));
+ assertEquals("20",
result.getEffectiveConfigs().get("effective.unhealthy.threshold.delta.commits"));
+ assertEquals("3",
result.getEffectiveConfigs().get("observed.mdt.delta.commits.since.compaction"));
+ assertEquals(COMPACTION_INSTANT,
result.getEffectiveConfigs().get("observed.last.mdt.compaction.instant"));
+ assertEquals("3",
result.getEffectiveConfigs().get("observed.lag.duration.minutes"));
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestMetadataTableSyncCheck.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestMetadataTableSyncCheck.java
new file mode 100644
index 000000000000..cb9aac462dc2
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestMetadataTableSyncCheck.java
@@ -0,0 +1,165 @@
+/*
+ * 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.utilities.health;
+
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.metadata.MetadataPartitionType;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests the skip paths of {@link MetadataTableSyncCheck}, which need no
metadata table and so no
+ * Spark session. The verdicts over a real metadata table are covered by
+ * {@link TestMetadataTableSyncCheckFunctional}.
+ */
+public class TestMetadataTableSyncCheck {
+
+ @TempDir
+ Path tempDir;
+
+ private String basePath;
+ private final MetadataTableSyncCheck check = new MetadataTableSyncCheck();
+
+ @BeforeEach
+ public void setUp() {
+ basePath = tempDir.resolve("table").toString();
+ }
+
+ private TypedProperties propsWithMetadataEnabled(boolean enabled) {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(HoodieMetadataConfig.ENABLE.key(),
String.valueOf(enabled));
+ return props;
+ }
+
+ private HealthCheckResult runCheck(HoodieTableMetaClient metaClient,
TypedProperties props) throws Exception {
+ return check.check(new
HealthCheckContext(HoodieTableMetaClient.reload(metaClient), props, false));
+ }
+
+ private HoodieTableMetaClient tableWithoutMetadataTable() throws Exception {
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath,
HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable.of(metaClient).addCommit("20260101100000000");
+ return metaClient;
+ }
+
+ @Test
+ public void tableWithoutMetadataTableIsSkipped() throws Exception {
+ // given a table that has never had a metadata table
+ HoodieTableMetaClient metaClient = tableWithoutMetadataTable();
+
+ // when the check runs with the metadata table enabled for the writer
+ HealthCheckResult result = runCheck(metaClient,
propsWithMetadataEnabled(true));
+
+ // then there is nothing to compare, and the check says so rather than
failing
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains("no metadata table"),
result.getSummary());
+ }
+
+ @Test
+ public void metadataTableDisabledForWriterIsSkipped() throws Exception {
+ // given any table
+ HoodieTableMetaClient metaClient = tableWithoutMetadataTable();
+
+ // when the writer is configured with the metadata table off
+ HealthCheckResult result = runCheck(metaClient,
propsWithMetadataEnabled(false));
+
+ // then the check does not apply, and the reason names the property
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+
assertTrue(result.getSummary().contains(HoodieMetadataConfig.ENABLE.key()),
result.getSummary());
+ }
+
+ @Test
+ public void missingEnablePropertySkipsNamingIt() throws Exception {
+ // given any table
+ HoodieTableMetaClient metaClient = tableWithoutMetadataTable();
+
+ // when the check runs without knowing whether the writer uses the
metadata table
+ HealthCheckResult result = runCheck(metaClient, new TypedProperties());
+
+ // then it declines to judge and says which property to supply
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+
assertTrue(result.getSummary().contains(HoodieMetadataConfig.ENABLE.key()),
result.getSummary());
+ assertTrue(result.getSummary().contains("--apply-all-defaults"),
result.getSummary());
+ }
+
+ @Test
+ public void smallTableSamplesEveryPartitionInSortedOrder() {
+ // given fewer partitions than the sample bound, listed in arbitrary order
+ List<String> partitions = Arrays.asList("2024/03/02", "2024/03/01",
"2024/03/03");
+
+ // when the sample is taken
+ List<String> sampled = MetadataTableSyncCheck.samplePartitions(partitions);
+
+ // then every partition is compared, in a stable order
+ assertEquals(Arrays.asList("2024/03/01", "2024/03/02", "2024/03/03"),
sampled);
+ }
+
+ @Test
+ public void largeTableSamplesTheNewestPartitionsPlusAFewOfTheOldest() {
+ // given a date-partitioned table with many more partitions than the
sample bound
+ List<String> partitions = new ArrayList<>();
+ for (int day = 1; day <= 60; day++) {
+ partitions.add(String.format("2024/03/%02d", day));
+ }
+ Collections.shuffle(partitions, new Random(7));
+
+ // when the sample is taken
+ List<String> sampled = MetadataTableSyncCheck.samplePartitions(partitions);
+
+ // then it is the most recently written partitions -- where drift
originates -- plus the oldest few for coverage
+ assertEquals(MetadataTableSyncCheck.MAX_PARTITIONS_TO_COMPARE +
MetadataTableSyncCheck.OLDEST_PARTITIONS_TO_INCLUDE, sampled.size());
+ assertEquals(Arrays.asList("2024/03/01", "2024/03/02"), sampled.subList(0,
2));
+ assertEquals("2024/03/41", sampled.get(2));
+ assertEquals("2024/03/60", sampled.get(sampled.size() - 1));
+ }
+
+ @Test
+ public void metadataTableRegisteredButMissingOnStorageIsSkipped() throws
Exception {
+ // given a table whose hoodie.properties claims a FILES partition, but
with no metadata table directory
+ HoodieTableMetaClient metaClient = tableWithoutMetadataTable();
+ Properties registration = new Properties();
+
registration.setProperty(HoodieTableConfig.TABLE_METADATA_PARTITIONS.key(),
MetadataPartitionType.FILES.getPartitionPath());
+ HoodieTableConfig.update(metaClient.getStorage(),
metaClient.getMetaPath(), registration);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient,
propsWithMetadataEnabled(true));
+
+ // then it skips with the path it looked for, instead of failing to open
the metadata table
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains("does not exist"),
result.getSummary());
+ }
+}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestMetadataTableSyncCheckFunctional.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestMetadataTableSyncCheckFunctional.java
new file mode 100644
index 000000000000..5e9208981c8b
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestMetadataTableSyncCheckFunctional.java
@@ -0,0 +1,253 @@
+/*
+ * 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.utilities.health;
+
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteClientTestUtils;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.fs.FSUtils;
+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.testutils.HoodieTestTable;
+import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.testutils.HoodieSparkClientTestBase;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.function.Predicate;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link MetadataTableSyncCheck} against a table with a real metadata
table, which needs a
+ * Spark write client to produce. The skip paths that need no metadata table
live in the lighter
+ * {@link TestMetadataTableSyncCheck}.
+ */
+public class TestMetadataTableSyncCheckFunctional extends
HoodieSparkClientTestBase {
+
+ private final MetadataTableSyncCheck check = new MetadataTableSyncCheck();
+
+ private TypedProperties propsWithMetadataEnabled() {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(HoodieMetadataConfig.ENABLE.key(), "true");
+ return props;
+ }
+
+ private HealthCheckResult runCheck() throws Exception {
+ return runCheck(propsWithMetadataEnabled());
+ }
+
+ private HealthCheckResult runCheck(TypedProperties props) throws Exception {
+ HoodieTableMetaClient freshMetaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf.newInstance())
+ .setBasePath(basePath)
+ .setLoadActiveTimelineOnLoad(true)
+ .build();
+ return check.check(new HealthCheckContext(freshMetaClient, props, false));
+ }
+
+ private HoodieWriteConfig writeConfigWithMetadataTable() {
+ return HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withParallelism(2, 2)
+ .withBulkInsertParallelism(2)
+ .forTable("health_check_table")
+
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.SIMPLE).build())
+
.withCompactionConfig(HoodieCompactionConfig.newBuilder().withInlineCompaction(false).build())
+
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(true).build())
+ .build();
+ }
+
+ /** A Merge-on-Read table with a metadata table, holding both base files and
log files. */
+ private void writeTableWithMetadataTable() throws Exception {
+ initMetaClient(HoodieTableType.MERGE_ON_READ);
+ try (SparkRDDWriteClient client =
getHoodieWriteClient(writeConfigWithMetadataTable())) {
+ String insertInstant = WriteClientTestUtils.createNewInstantTime();
+ List<HoodieRecord> inserts = dataGen.generateInserts(insertInstant, 30);
+ WriteClientTestUtils.startCommitWithTime(client, insertInstant);
+ JavaRDD<WriteStatus> insertStatuses =
client.bulkInsert(jsc.parallelize(inserts, 1), insertInstant);
+ client.commit(insertInstant, insertStatuses);
+
+ String updateInstant = WriteClientTestUtils.createNewInstantTime();
+ List<HoodieRecord> updates = dataGen.generateUpdates(updateInstant, 15);
+ WriteClientTestUtils.startCommitWithTime(client, updateInstant);
+ JavaRDD<WriteStatus> updateStatuses =
client.upsert(jsc.parallelize(updates, 1), updateInstant);
+ client.commit(updateInstant, updateStatuses);
+ }
+ }
+
+ /** Removes one file of the given kind directly on storage, bypassing the
timeline and the metadata table. */
+ private StoragePath deleteOneFileFromStorage(String partition,
Predicate<StoragePath> kind) throws Exception {
+ List<StoragePathInfo> entries = storage.listDirectEntries(new
StoragePath(basePath, partition));
+ StoragePathInfo victim = entries.stream()
+ .filter(e -> kind.test(e.getPath()))
+ .findFirst()
+ .orElseThrow(() -> new IllegalStateException("no matching file in " +
partition));
+ assertTrue(storage.deleteFile(victim.getPath()));
+ return victim.getPath();
+ }
+
+ private void assertInconsistencyNames(HealthCheckResult result, String
partition, StoragePath file, String fileKind) {
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertEquals("1",
result.getEffectiveConfigs().get("observed.partitions.inconsistent"));
+ assertTrue(result.getFindings().stream().anyMatch(
+ f -> f.contains("'" + partition + "'") &&
f.contains(file.getName()) && f.contains(fileKind + " files listed by the
metadata table")),
+ "expected the partition and " + fileKind + " file to be named, got: "
+ result.getFindings());
+ assertTrue(result.getFindings().stream().anyMatch(f ->
f.contains("HoodieMetadataTableValidator")),
+ "expected a pointer to the full validator, got: " +
result.getFindings());
+ }
+
+ @Test
+ public void caughtUpTableWithConsistentListingIsHealthy() throws Exception {
+ // given a table whose every write was mirrored into the metadata table
+ writeTableWithMetadataTable();
+
+ // when the check runs
+ HealthCheckResult result = runCheck();
+
+ // then the metadata table is synced and agrees with storage in every
sampled partition
+ assertEquals(HealthStatus.HEALTHY, result.getStatus(), result.getSummary()
+ " " + result.getFindings());
+ assertEquals("0",
result.getEffectiveConfigs().get("observed.data.table.writes.behind"));
+ assertEquals("0",
result.getEffectiveConfigs().get("observed.partitions.inconsistent"));
+
assertEquals(result.getEffectiveConfigs().get("observed.data.table.latest.completed.instant"),
+ result.getEffectiveConfigs().get("observed.mdt.synced.instant"));
+
assertTrue(Integer.parseInt(result.getEffectiveConfigs().get("observed.partitions.sampled"))
> 0);
+ }
+
+ @Test
+ public void baseFileMissingFromStorageIsUnhealthyNamingThePartitionAndFile()
throws Exception {
+ // given a consistent table, with one base file then removed directly on
storage
+ writeTableWithMetadataTable();
+ StoragePath deleted =
deleteOneFileFromStorage(DEFAULT_FIRST_PARTITION_PATH, FSUtils::isBaseFile);
+
+ // when the check runs
+ HealthCheckResult result = runCheck();
+
+ // then the metadata table and storage disagree, and the finding says
where and on what
+ assertInconsistencyNames(result, DEFAULT_FIRST_PARTITION_PATH, deleted,
"base");
+ }
+
+ @Test
+ public void logFileMissingFromStorageIsUnhealthyNamingThePartitionAndFile()
throws Exception {
+ // given a consistent table, with one log file then removed directly on
storage
+ writeTableWithMetadataTable();
+ StoragePath deleted =
deleteOneFileFromStorage(DEFAULT_FIRST_PARTITION_PATH, FSUtils::isLogFile);
+
+ // when the check runs
+ HealthCheckResult result = runCheck();
+
+ // then log files are compared too, not only base files
+ assertInconsistencyNames(result, DEFAULT_FIRST_PARTITION_PATH, deleted,
"log");
+ }
+
+ @Test
+ public void
dataTableWriteMissingFromMetadataTableIsUnhealthyAndNotTreatedAsSelfHealing()
throws Exception {
+ // given a consistent table, then a completed data-table write the
metadata table never saw
+ writeTableWithMetadataTable();
+ HoodieTableMetaClient dataMetaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf.newInstance()).setBasePath(basePath).build();
+ String unmirroredInstant = dataMetaClient.createNewInstantTime(false);
+ HoodieTestTable.of(dataMetaClient).addCommit(unmirroredInstant);
+
+ // when the check runs
+ HealthCheckResult result = runCheck();
+
+ // then the lag is reported, and the operator is told not to wait it out:
a later write does
+ // not replay the missed instant, it only moves the synced instant past
the gap
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertEquals("1",
result.getEffectiveConfigs().get("observed.data.table.writes.behind"));
+ assertEquals(unmirroredInstant,
result.getEffectiveConfigs().get("observed.data.table.latest.completed.instant"));
+ assertEquals("0",
result.getEffectiveConfigs().get("observed.partitions.inconsistent"));
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("Do not
wait for the lag to clear")),
+ "expected the finding to say lag does not clear itself, got: " +
result.getFindings());
+ assertTrue(result.getFindings().stream().anyMatch(f ->
f.contains("HoodieMetadataTableValidator")),
+ "lag with no in-flight write is a possible inconsistency, so the
validator is the next step: "
+ + result.getFindings());
+ }
+
+ @Test
+ public void metadataTableLeftOnStorageWithWriterDisabledIsUnhealthy() throws
Exception {
+ // given a real metadata table on storage, and a writer configured with it
off --
+ // what happens whenever hoodie.metadata.auto.delete.partitions is false
+ writeTableWithMetadataTable();
+ TypedProperties disabled = new TypedProperties();
+ disabled.setProperty(HoodieMetadataConfig.ENABLE.key(), "false");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(disabled);
+
+ // then it does not skip: nothing maintains the table while readers still
use it
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus(),
+ "a metadata table nothing maintains is the risky case, not an
uninteresting one");
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("Readers
are not bound")),
+ "expected the reader consequence to be named, got: " +
result.getFindings());
+ }
+
+ @Test
+ public void rebuildRemedyDoesNotRecommendTheDisableReEnableDance() throws
Exception {
+ // given a metadata table whose listing has drifted from storage
+ writeTableWithMetadataTable();
+ deleteOneFileFromStorage("2015/03/16", path ->
path.getName().endsWith(".parquet"));
+
+ // when the check reports the inconsistency
+ HealthCheckResult result = runCheck();
+
+ // then the remedy is explicit about what it does, and warns off the
config side effect
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getFindings().stream().anyMatch(f ->
f.contains("metadata delete")
+ && f.contains("metadata create")),
+ "expected the hudi-cli remedy, got: " + result.getFindings());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("drops
every index")),
+ "expected the dropped indexes to be named, got: " +
result.getFindings());
+ }
+
+ @Test
+ public void validatorCommandCarriesTheSchemeQualifiedBasePath() throws
Exception {
+ // given a drifted listing, so the finding that recommends the validator
is emitted
+ writeTableWithMetadataTable();
+ deleteOneFileFromStorage("2015/03/16", path ->
path.getName().endsWith(".parquet"));
+
+ // when the check runs
+ HealthCheckResult result = runCheck();
+
+ // then the recommended command names a scheme-qualified path.
HoodieMetadataTableValidator
+ // compares the two listings by full path string: the metadata side echoes
the path it was
+ // given, the file-system side is always returned scheme-qualified, so a
bare --base-path makes
+ // every file compare unequal and the validator reports drift on a healthy
table.
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getFindings().stream().anyMatch(f ->
f.contains("--base-path file:")),
+ "expected a scheme-qualified --base-path in the validator command,
got: " + result.getFindings());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("The
scheme matters")),
+ "expected the finding to say why the scheme matters, got: " +
result.getFindings());
+ }
+}
