You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by co...@apache.org on 2026/10/16 09:33:21 UTC
(hudi) branch master updated: feat(utilities): add HoodieTableHealthChecker for table service health (#20151)
This is an automated email from the ASF dual-hosted git repository.
codope 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 5764e96b68dd feat(utilities): add HoodieTableHealthChecker for table service health (#20151)
5764e96b68dd is described below
commit 5764e96b68dd8d76dcf1d1b74491f382184f067c
Author: Sivabalan Narayanan <n....@gmail.com>
AuthorDate: Fri Oct 16 02:33:14 2026 -0700
feat(utilities): add HoodieTableHealthChecker for table service health (#20151)
* feat(utilities): add HoodieTableHealthChecker for table service health
Adds a read-only utility that reports whether a Hudi table's table
services are keeping up with ingestion, meant for a scheduled run whose
exit code feeds alerting. Four timeline-only checks ship in this change:
compaction cadence, cleaner cadence, savepoints blocking archival, and
archival cadence. A small framework (HealthCheck, HealthCheckResult,
HealthCheckContext, HealthStatus) lets later checks plug in.
Verdicts are relative to how the table is written, so writer properties
are required via --props / --hoodie-conf. A check whose governing
property was not supplied reports SKIPPED and names it rather than
guessing against Hudi defaults; --apply-all-defaults opts into defaults
explicitly. Every check echoes the effective configuration it used.
Output is a human table or --output JSON, with per-check status,
summary, actionable findings, and effective configs, so an agent can
consume it. A companion Claude Code skill, hudi-ai-operator/skills/
hudi-table-health, interprets the report and recommends next steps
without acting on the table.
No new hoodie.* configs, no public API or storage format changes.
* fix(utilities): model the retention the archiver actually enforces
Review feedback on the health checker, with reproductions.
hoodie.keep.max.commits is not on its own the bound archival enforces.
ArchivalUtils.getMinAndMaxInstantsToKeep raises both bounds whenever the
cleaner retains more history than archival would, because archival must not
move an instant off the active timeline while the cleaner still considers its
files live. The archival check compared against the raw configured maximum, so
a supported setup -- clean.commits.retained=100 against keep.max.commits=30 --
came out UNHEALTHY, and KEEP_LATEST_BY_HOURS with an hour of retention did the
same at 50 commits. The check now reproduces that calculation from the cleaner
configuration and reports the effective bound it derived. It cannot call
ArchivalUtils directly: that needs a HoodieTable, and this tool reads only the
timeline.
It was also counting the wrong thing. getWriteTimeline() includes requested and
inflight compaction and clustering, which archival has not had the chance to
consider; the archiver reasons over completed commits, and so does the check
now.
Attribution is better where it can be. Archival never moves past the earliest
pending write instant, so when one is on the timeline the finding names it --
that is the thing an operator can act on. Compaction retention and the metadata
table's own compaction also hold archival back and are not visible from here,
so the closing finding says so rather than implying a savepoint is the only
cause.
The cleaner check blamed the wrong config. With keep.min.commits at or below
clean.commits.retained, archival trims the timeline before the cleaner ever has
enough commits to act on, so no clean is ever scheduled -- 60 file versions and
zero cleans, with the finding pointing at hoodie.clean.automatic. That pairing
is now detected and named, and the generic finding is kept for when the configs
do not explain it.
Exit codes: 1 is what the JVM returns when the process fails to start, so a
classpath mistake read as UNHEALTHY. Unhealthy moves to 3, a check that could
not run to 4, and unhealthy now outranks a failed check rather than the other
way round -- the table's verdict is the more actionable of the two.
The documented java -cp invocation could not work: the bundle does not carry
slf4j, Hadoop or Jackson. Replaced with spark-submit, including in the cron
example, which had the same problem and keyed off the old exit codes. Renamed
HoodieTableHealthChecker.md to README.md to match the convention.
Found while fixing the above: the cleaner's earliest-commit-to-retain was
resolved by parsing the latest commit's instant time unconditionally, but only
KEEP_LATEST_BY_HOURS needs it as wall-clock. An instant time that would not
parse silently disabled the whole derivation and fell back to the configured
bound.
* fix(utilities): do not judge the cleaner on an append-only table
A table that only ever receives inserts never produces a clean instant, and
that is correct: cleaning retires superseded file slices, and an append-only
workload never supersedes anything. Every file group has a single slice and
nothing is replaced, so the cleaner runs and finds no work.
The check read an empty cleaner timeline as a stalled service, so such a table
was reported UNHEALTHY every day with no action available to the operator --
the kind of standing false alarm that teaches people to ignore the tool.
Append-only is now read from the commit metadata rather than from
hoodie.datasource.write.operation, because what the table has actually
received is the question: a writer configured for upsert that has only ever
inserted still has nothing to clean, and a property file can be stale or
absent. One insert_overwrite, upsert or delete anywhere on the timeline means
file slices can be superseded and the normal cadence judgement applies again.
Conservative on failure. An unreadable commit, or an operation type the
metadata does not carry, means the exemption does not apply and the check
judges as before -- suppressing a real finding is the worse error of the two,
so the exemption has to be earned by every completed write on the timeline.
* feat(utilities): give health checks an optional Spark context
Checks so far read only the timeline, so the checker has never needed Spark.
Two things change that. The metadata-table sync check should delegate its
listing comparison to HoodieMetadataTableValidator rather than reimplement a
narrower one, and the validator takes a JavaSparkContext. And the compaction
and cleaner checks walk file slices per file group and file versions per
partition on the driver, which does not hold on a table with hundreds of
thousands of file groups.
HealthCheckContext now carries an Option<JavaSparkContext>, and the driver an
optional one it passes through. Both keep their existing constructors, so a
caller that does not supply a context gets exactly what it got before, and the
checks that only read the timeline stay runnable without a cluster. A check
that needs a context and is not given one reports SKIPPED naming what it
needed, rather than failing the run -- the same contract the checker already
applies to a missing writer property.
No check uses it yet; this is the plumbing.
* feat(utilities): build the Spark context in the driver
Follows the previous commit, which added the optional context but left main
passing nothing, so a check needing one would always have found it absent.
main now builds it unconditionally and stops it in a finally. Building it only
for the checks that need one would have made the selected check set decide
whether the tool can start, which is a worse property than paying for a local
context on a run that does not use it. --spark-master sets the master when the
job does not already have one; under spark-submit the submitted master wins.
The README claimed no Spark session was used. That was true when every check
read only the timeline and is about to stop being true, so it now says to run
under spark-submit and that listing comparisons scan the table while the
timeline checks do not.
---
hudi-ai-operator/README.md | 94 ++++++
.../skills/hudi-table-health/README.md | 60 ++++
hudi-ai-operator/skills/hudi-table-health/SKILL.md | 108 ++++++
.../skills/hudi-table-health/references/checks.md | 90 +++++
.../hudi/utilities/HoodieTableHealthChecker.java | 370 +++++++++++++++++++++
.../main/java/org/apache/hudi/utilities/README.md | 225 +++++++++++++
.../utilities/health/ArchivalCadenceCheck.java | 260 +++++++++++++++
.../hudi/utilities/health/CleanerCadenceCheck.java | 245 ++++++++++++++
.../utilities/health/CompactionCadenceCheck.java | 206 ++++++++++++
.../apache/hudi/utilities/health/HealthCheck.java | 53 +++
.../hudi/utilities/health/HealthCheckContext.java | 136 ++++++++
.../hudi/utilities/health/HealthCheckResult.java | 117 +++++++
.../apache/hudi/utilities/health/HealthStatus.java | 48 +++
.../health/SavepointArchivalBlockCheck.java | 150 +++++++++
.../utilities/TestHoodieTableHealthChecker.java | 253 ++++++++++++++
.../utilities/health/TestArchivalCadenceCheck.java | 185 +++++++++++
.../utilities/health/TestCleanerCadenceCheck.java | 329 ++++++++++++++++++
.../health/TestCompactionCadenceCheck.java | 213 ++++++++++++
.../utilities/health/TestHealthCheckContext.java | 103 ++++++
.../hudi/utilities/health/TestHealthStatus.java | 62 ++++
.../health/TestSavepointArchivalBlockCheck.java | 190 +++++++++++
21 files changed, 3497 insertions(+)
diff --git a/hudi-ai-operator/README.md b/hudi-ai-operator/README.md
new file mode 100644
index 000000000000..df694f9272fa
--- /dev/null
+++ b/hudi-ai-operator/README.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.
+-->
+
+# Hudi AI Operator
+
+Claude Code Skills for the people who deploy, operate and manage Apache Hudi tables: the ones who
+own the writer properties, get paged when ingestion stalls, and have to decide whether a thing is
+broken or merely slow.
+
+This is the counterpart to `hudi-agent-gateway`, which serves the other persona — people who
+*consume* data from Hudi tables, through SQL, a chat interface, or an MCP endpoint. That side cares
+about querying. This side cares about whether the table underneath is healthy, laid out well, and
+configured the way its owner thinks it is.
+
+## What is here
+
+| Skill | Question it answers | What it drives |
+|---|---|---|
+| [`hudi-table-health`](skills/hudi-table-health/README.md) | Are the table services keeping up? | `HoodieTableHealthChecker` |
+| [`hudi-mdt-health`](skills/hudi-mdt-health/README.md) | Is the metadata table caught up and trustworthy? | `HoodieTableHealthChecker` (metadata checks) |
+| [`hudi-table-layout`](skills/hudi-table-layout/README.md) | Is the data laid out well? | `HoodieTableLayoutAnalyzer` |
+| [`hudi-rca`](skills/hudi-rca/README.md) | Why did this break? | A catalog of known failure patterns |
+| [`hudi-config-consultant`](skills/hudi-config-consultant/README.md) | What does this config actually do? | A catalog generated by `scripts/generate_config_catalog.py` |
+| [`hudi-optimizer`](skills/hudi-optimizer/README.md) | This works — could it be better? | `HoodieSparkEventLogAnalyzer`, plus the health and layout tools |
+
+The tools live in `hudi-utilities`; each has its own `.md` runbook beside the class. The Skills are
+the interpretation layer on top: they choose the flags, read the JSON, and explain the verdict.
+
+## How the pieces fit
+
+The split is by question, and the questions do not overlap much:
+
+- **Health** is about cadence. Compaction, cleaning, archival, savepoints, metadata sync — is each
+ service running often enough for the write rate it is behind?
+- **Layout** is about shape. How bytes and files spread across partitions, whether the table is
+ micro-partitioned or piling up small files, which partitions are hot.
+- **RCA** is about failure. A job died; walk the `Caused by` chain, classify it, and give an ordered
+ mitigation ladder.
+- **Config consultant** is about reading. Not the key and its default, which the docs already give,
+ but where the value is read, what gates that read, and what is read alongside it.
+- **Optimizer** is about the middle case: the job runs, it is correct, and it costs more than it
+ needs to.
+
+Several Skills deliberately **delegate detection to the tools rather than re-implementing it**. The
+optimizer runs the health checker and the layout analyzer rather than carrying its own copy of those
+heuristics; the mdt-health Skill drives the same `HoodieTableHealthChecker` binary as table-health,
+narrowed to the metadata checks. One detection implementation, in Java, with tests — the Skills add
+judgement on top of it, and a fix to a detector reaches every Skill at once.
+
+Skills also hand off to each other by name. `hudi-optimizer` stops at anything that would change
+what is on storage — table type, index type, partitioning — and hands the measurements to
+`hudi-architect` (in `hudi-agent-gateway/skills/hudi-architect`) for an Architecture Decision
+Record. If the job is failing rather than slow, that is `hudi-rca`.
+
+## Install
+
+Each Skill is a self-contained directory. Copy the one you want:
+
+```bash
+# From the root of a Hudi checkout.
+SKILL=hudi-ai-operator/skills/hudi-table-health
+
+# user-level (available in every project)
+mkdir -p ~/.claude/skills && cp -r "$SKILL" ~/.claude/skills/
+# or project-level
+mkdir -p .claude/skills && cp -r "$SKILL" .claude/skills/
+```
+
+Then invoke it by name in Claude Code — `/hudi-table-health`, `/hudi-rca`, and so on.
+
+Most of these Skills need the `hudi-utilities-bundle` jar from the same Hudi release, reachable from
+wherever Claude Code runs commands, plus read access to the table's base path. The tools are
+read-only and never modify the table. `hudi-config-consultant` is the exception: it needs no jar, no
+cluster and no table, because it reads only its catalog file.
+
+## Per-Skill detail
+
+Each Skill directory has its own `README.md` covering what it does, what it is not, its exact
+requirements, and its known limits. Start there rather than here — this file is a map, not a manual.
diff --git a/hudi-ai-operator/skills/hudi-table-health/README.md b/hudi-ai-operator/skills/hudi-table-health/README.md
new file mode 100644
index 000000000000..106e9043168f
--- /dev/null
+++ b/hudi-ai-operator/skills/hudi-table-health/README.md
@@ -0,0 +1,60 @@
+<!--
+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 Table Health
+
+A Claude Code Skill that runs `HoodieTableHealthChecker` against a Hudi table and turns its JSON
+report into a diagnosis: which table services are keeping up, which are not, and what to do next.
+
+It pairs with the utility of the same name 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 thing that makes verdicts wrong: it
+insists on the writer's properties rather than silently judging the table against Hudi defaults.
+
+## Install
+
+```bash
+SKILL=hudi-ai-operator/skills/hudi-table-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-table-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. The tool is read-only and never modifies the 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 table-service cadence: compaction, cleaning, archival, savepoints. Data layout
+(small files, micro-partitions) and metadata-table consistency are separate tools with their own
+Skills.
diff --git a/hudi-ai-operator/skills/hudi-table-health/SKILL.md b/hudi-ai-operator/skills/hudi-table-health/SKILL.md
new file mode 100644
index 000000000000..b5e35ee68093
--- /dev/null
+++ b/hudi-ai-operator/skills/hudi-table-health/SKILL.md
@@ -0,0 +1,108 @@
+---
+name: hudi-table-health
+description: Checks whether an Apache Hudi table's table services (compaction, cleaning, archival, savepoints) are keeping up with ingestion, by running the HoodieTableHealthChecker utility and interpreting its JSON report. Invoke when a user asks whether a Hudi table is healthy, whether compaction or cleaning is falling behind, why the active timeline keeps growing, or what to do about a savepoint blocking archival.
+---
+
+<!--
+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 Table Health
+
+You are helping an operator find out whether a Hudi table is progressing as expected, and what to
+do if it is not. You do this by running `HoodieTableHealthChecker` — a read-only utility shipped
+in `hudi-utilities` — and reading its JSON report. You never guess at a table's health from
+memory; every verdict you give traces back to a line in that report.
+
+## What the tool answers
+
+| Check | Question |
+|---|---|
+| `compaction` | Are delta commits piling up past the configured trigger on a Merge-on-Read table? |
+| `cleaner` | Are write commits piling up without a clean completing? |
+| `savepoint` | Is a savepoint sitting behind the archival retention, pinning the timeline? |
+| `archival` | Is the active timeline staying bounded? |
+
+Each check reports `HEALTHY`, `UNHEALTHY`, or `SKIPPED`. `SKIPPED` means the check could not
+judge — usually because a writer property it 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.** Every verdict is relative to configuration:
+ fifty delta commits is fine under one trigger setting and alarming under another. Ask for a
+ properties file or the individual `hoodie.*` values. The properties that matter per check are
+ listed in `references/checks.md`.
+
+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, so 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. The checks are timeline-only, 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> \
+ --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.
+
+Use `--checks compaction,cleaner` to narrow the run when the user is investigating one service.
+
+## 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; do not paraphrase them into something vaguer. Then add the remedy from
+ `references/checks.md` for that check.
+- **`SKIPPED`** — the `summary` names the missing property. Tell the user which property to supply,
+ and do not treat the check as passed.
+- **`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
+("37 commits since the last clean at 20260901120000000").
+
+## Step 4 — recommend, don't act
+
+This skill diagnoses. It does not run compaction, delete savepoints, 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. Deleting a savepoint
+is not.
+
+## Guardrails
+
+- Every claim about the table's health must come from the report. If you have not run the tool,
+ say you have not, and do not speculate.
+- Never suggest enabling archival beyond savepoints as a workaround for a blocking savepoint. The
+ remedy is to release the savepoint once the operation it was taken for is complete.
+- Never pass `--apply-all-defaults` without the user's explicit choice.
+- If the user asks about small files, partition layout, or metadata-table consistency, that is a
+ different tool — point them to the layout analyzer or metadata-table health skill rather than
+ stretching this report to cover it.
diff --git a/hudi-ai-operator/skills/hudi-table-health/references/checks.md b/hudi-ai-operator/skills/hudi-table-health/references/checks.md
new file mode 100644
index 000000000000..6c3e246f253c
--- /dev/null
+++ b/hudi-ai-operator/skills/hudi-table-health/references/checks.md
@@ -0,0 +1,90 @@
+<!--
+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": "compaction",
+ "status": "UNHEALTHY",
+ "summary": "one line for the operator",
+ "findings": ["actionable sentence", "..."],
+ "effectiveConfigs": {
+ "hoodie.compact.inline.max.delta.commits": "5",
+ "effective.unhealthy.threshold.delta.commits": "10",
+ "observed.delta.commits.since.last.compaction": "27"
+ }
+ }
+ ],
+ "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`).
+
+## Per check
+
+### compaction
+
+| | |
+|---|---|
+| Applies to | Merge-on-Read only; `SKIPPED` on Copy-on-Write |
+| Needs | `hoodie.compact.inline.trigger.strategy`, `hoodie.compact.inline.max.delta.commits` (and `hoodie.compact.inline.max.delta.seconds` for time-based strategies) |
+| Unhealthy when | delta commits since the last completed compaction exceed 2x what the trigger allows |
+| Finding "No compaction is scheduled" | The scheduler is not running or compaction is disabled. Check `hoodie.compact.inline` / `hoodie.compact.schedule.inline` on the writer, or the async compaction job. |
+| Finding "scheduled but not executed" | Plans exist on the timeline but never complete. Look at the async compactor's logs for repeated failures; `compactions show all` in `hudi-cli` lists pending plans. |
+| Remedy | Get the compactor running; if a plan is corrupt, `compaction validate` then `compaction unschedule` in `hudi-cli`. Running compaction is reversible in the sense that it does not lose data; unscheduling a plan discards only the plan. |
+
+### cleaner
+
+| | |
+|---|---|
+| Needs | `hoodie.clean.trigger.max.commits` (echoes `hoodie.clean.policy`, `hoodie.clean.commits.retained` if given) |
+| Unhealthy when | write commits since the last completed clean exceed max(2x trigger, 5) |
+| Finding "No clean is scheduled" | Cleaning is disabled or the job is not running. Check `hoodie.clean.automatic` and `hoodie.clean.async`. |
+| Finding "has not completed" | A clean is requested or inflight and stuck. Under concurrent writers a clean that keeps losing the table lock is the usual cause; check the writer logs. |
+| Remedy | `cleans run` in `hudi-cli` runs one clean now. Cleaning is not reversible — it deletes file versions — but it only deletes what the retention policy already permits. |
+
+### savepoint
+
+| | |
+|---|---|
+| Needs | `hoodie.keep.max.commits` |
+| Unhealthy when | a savepoint sits further back than the archival retention window |
+| Also flags | any savepoint older than 7 days, even when not yet blocking, as likely forgotten |
+| Why it matters | Archival stops at the earliest savepoint. The active timeline cannot shrink past it until the savepoint is released. |
+| Remedy | Once the operation the savepoint was taken for is complete: `savepoint delete --commit <instant>` in `hudi-cli`. **Not reversible.** Confirm with the owner before recommending it, and never suggest configuring archival to skip savepoints instead. |
+
+### archival
+
+| | |
+|---|---|
+| Needs | `hoodie.keep.max.commits` |
+| Unhealthy when | write instants exceed 1.1x the configured maximum, or completed instants exceed 5000 regardless of configuration |
+| First thing to check | The `savepoint` check — a blocking savepoint is the most common cause. |
+| Otherwise | Confirm archival is enabled and look for repeated archival failures in the writer logs. `timeline show active` in `hudi-cli` shows the current size. |
+| Remedy | Release the blocking savepoint if there is one; otherwise get the writer's archival running. Archival is reversible in effect — archived instants remain readable from the archived timeline. |
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
new file mode 100644
index 000000000000..fbda47f4a776
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java
@@ -0,0 +1,370 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.utilities.health.ArchivalCadenceCheck;
+import org.apache.hudi.utilities.health.CleanerCadenceCheck;
+import org.apache.hudi.utilities.health.CompactionCadenceCheck;
+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.SavepointArchivalBlockCheck;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.spark.api.java.JavaSparkContext;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Reports whether a Hudi table's table services are keeping up with ingestion.
+ *
+ * <p>Intended to be run on a schedule -- daily is typical -- against tables an operator is
+ * responsible for. Each check answers one question about whether a service is progressing as
+ * expected, and the process exit code makes the overall verdict usable from a cron job or CI
+ * pipeline without parsing the output.
+ *
+ * <p>This tool is strictly read-only. It inspects timeline state and reports; it never schedules or
+ * runs a table service, and never writes to the table.
+ *
+ * <h3>Writer properties are required</h3>
+ *
+ * <p>Nearly every question this tool answers is relative to how the table is written. Whether
+ * fifty delta commits since the last compaction is fine or alarming depends on the configured
+ * trigger; whether a thousand instants in the active timeline is fine depends on the configured
+ * retention. Hudi has a default for each of these, so the tool could always produce an answer --
+ * but for a table written with non-default settings that answer would be wrong, and wrong in the
+ * direction that erodes trust, reporting healthy tables as unhealthy.
+ *
+ * <p>So by default a check whose governing property was not supplied reports {@code SKIPPED} and
+ * names the property, rather than guessing. Supply properties with {@code --props} or
+ * {@code --hoodie-conf}, or pass {@code --apply-all-defaults} to explicitly accept Hudi defaults
+ * for anything unset.
+ *
+ * <h3>Exit codes</h3>
+ * <ul>
+ * <li>{@code 0} -- every check that ran reported healthy</li>
+ * <li>{@code 2} -- bad arguments, or the table could not be read</li>
+ * <li>{@code 3} -- at least one check reported unhealthy</li>
+ * <li>{@code 4} -- no check reported unhealthy, but at least one could not run</li>
+ * </ul>
+ *
+ * <p>{@code 1} is deliberately unused: the JVM returns it when the process fails to start at all,
+ * so sharing it would make a classpath mistake read as a finding about the table.
+ *
+ * <h3>Example</h3>
+ * <pre>
+ * spark-submit --master local \
+ * --class org.apache.hudi.utilities.HoodieTableHealthChecker \
+ * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.12-1.3.0-SNAPSHOT.jar \
+ * --base-path s3://bucket/path/to/table \
+ * --props s3://bucket/path/to/writer.properties \
+ * --output JSON
+ * </pre>
+ */
+@Slf4j
+public class HoodieTableHealthChecker implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ static final int EXIT_HEALTHY = 0;
+ /**
+ * Reserved. The JVM itself exits 1 when the process fails to start -- a missing class, a bad
+ * classpath -- so a verdict must not share that code, or a broken invocation reads as a finding
+ * about the table.
+ */
+ static final int EXIT_ERROR = 2;
+ /** At least one check reported UNHEALTHY. Outranks {@link #EXIT_CHECK_ERRORED}. */
+ static final int EXIT_UNHEALTHY = 3;
+ /** No check reported UNHEALTHY, but at least one could not run. */
+ static final int EXIT_CHECK_ERRORED = 4;
+
+ private static final ObjectMapper JSON_MAPPER =
+ new ObjectMapper().enable(SerializationFeature.INDENT_OUTPUT);
+
+ private final Config cfg;
+ private final TypedProperties props;
+ private final HoodieTableMetaClient metaClient;
+ private final Option<JavaSparkContext> jsc;
+
+ public HoodieTableHealthChecker(Configuration hadoopConf, Config cfg) {
+ this(hadoopConf, cfg, Option.empty());
+ }
+
+ /**
+ * @param jsc a Spark context for checks that need one. Checks that only read the timeline do not,
+ * and the tool stays usable without one; a check that needs it and is not given it
+ * reports SKIPPED rather than failing the run.
+ */
+ public HoodieTableHealthChecker(Configuration hadoopConf, Config cfg, Option<JavaSparkContext> jsc) {
+ this.cfg = cfg;
+ this.jsc = jsc;
+ this.props = UtilHelpers.buildProperties(hadoopConf, cfg.propsFilePath, cfg.configs);
+ this.metaClient = HoodieTableMetaClient.builder()
+ .setConf(HadoopFSUtils.getStorageConfWithCopy(hadoopConf))
+ .setBasePath(cfg.basePath)
+ .setLoadActiveTimelineOnLoad(true)
+ .build();
+ }
+
+ /** All checks this tool knows about, in report order. */
+ static List<HealthCheck> allChecks() {
+ return Arrays.asList(
+ new CompactionCadenceCheck(),
+ new CleanerCadenceCheck(),
+ new SavepointArchivalBlockCheck(),
+ new ArchivalCadenceCheck());
+ }
+
+ /**
+ * Runs the selected checks and renders the report.
+ *
+ * @return the exit code the process should terminate with.
+ */
+ public int run() {
+ List<HealthCheck> selected = selectChecks(cfg.checks);
+ HealthCheckContext context = new HealthCheckContext(metaClient, props, cfg.applyAllDefaults, jsc);
+
+ List<HealthCheckResult> results = new ArrayList<>();
+ List<String> erroredChecks = new ArrayList<>();
+
+ for (HealthCheck check : selected) {
+ try {
+ results.add(check.check(context));
+ } catch (Exception e) {
+ // One check failing to reach a verdict must not hide the others. Record it and continue.
+ log.error("Health check '{}' failed to run for table {}", check.getName(), cfg.basePath, e);
+ erroredChecks.add(check.getName() + ": " + e.getMessage());
+ }
+ }
+
+ if (OutputFormat.JSON == cfg.output) {
+ System.out.println(renderJson(results, erroredChecks));
+ } else {
+ System.out.println(renderTable(results, erroredChecks));
+ }
+
+ // An unhealthy table is the more actionable finding, so it outranks a check that could not
+ // run. Both are still reported in full in the output above.
+ if (results.stream().anyMatch(r -> r.getStatus().isUnhealthy())) {
+ return EXIT_UNHEALTHY;
+ }
+ if (!erroredChecks.isEmpty()) {
+ return EXIT_CHECK_ERRORED;
+ }
+ return EXIT_HEALTHY;
+ }
+
+ private List<HealthCheck> selectChecks(String requested) {
+ List<HealthCheck> available = allChecks();
+ if (StringUtils.isNullOrEmpty(requested) || "all".equalsIgnoreCase(requested)) {
+ return available;
+ }
+
+ Map<String, HealthCheck> byName = new LinkedHashMap<>();
+ available.forEach(c -> byName.put(c.getName(), c));
+
+ List<HealthCheck> selected = new ArrayList<>();
+ for (String name : requested.split(",")) {
+ String trimmed = name.trim().toLowerCase();
+ if (!byName.containsKey(trimmed)) {
+ throw new IllegalArgumentException(String.format(
+ "Unknown check '%s'. Available checks: %s", trimmed, String.join(", ", byName.keySet())));
+ }
+ selected.add(byName.get(trimmed));
+ }
+ return selected;
+ }
+
+ private HealthStatus overallStatus(List<HealthCheckResult> results) {
+ return results.stream()
+ .map(HealthCheckResult::getStatus)
+ .reduce(HealthStatus.SKIPPED, HealthStatus::max);
+ }
+
+ private String renderTable(List<HealthCheckResult> results, List<String> erroredChecks) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("\nHudi table health report\n");
+ sb.append(" table : ").append(cfg.basePath).append('\n');
+ sb.append(" type : ").append(metaClient.getTableType()).append('\n');
+ sb.append(" overall : ").append(overallStatus(results)).append("\n\n");
+
+ for (HealthCheckResult result : results) {
+ sb.append(String.format("[%-9s] %s%n", result.getStatus(), result.getCheckName()));
+ sb.append(" ").append(result.getSummary()).append('\n');
+
+ for (String finding : result.getFindings()) {
+ sb.append(" - ").append(finding).append('\n');
+ }
+
+ if (cfg.verbose && !result.getEffectiveConfigs().isEmpty()) {
+ sb.append(" config used:\n");
+ result.getEffectiveConfigs().forEach((k, v) -> sb.append(" ").append(k).append(" = ").append(v).append('\n'));
+ }
+ sb.append('\n');
+ }
+
+ if (!erroredChecks.isEmpty()) {
+ sb.append("Checks that could not run:\n");
+ erroredChecks.forEach(e -> sb.append(" - ").append(e).append('\n'));
+ }
+
+ return sb.toString();
+ }
+
+ private String renderJson(List<HealthCheckResult> results, List<String> erroredChecks) {
+ ObjectNode root = JSON_MAPPER.createObjectNode();
+ root.put("basePath", cfg.basePath);
+ root.put("tableType", metaClient.getTableType().name());
+ root.put("overallStatus", overallStatus(results).name());
+
+ ArrayNode checks = root.putArray("checks");
+ for (HealthCheckResult result : results) {
+ ObjectNode node = checks.addObject();
+ node.put("name", result.getCheckName());
+ node.put("status", result.getStatus().name());
+ node.put("summary", result.getSummary());
+
+ ArrayNode findings = node.putArray("findings");
+ result.getFindings().forEach(findings::add);
+
+ ObjectNode configs = node.putObject("effectiveConfigs");
+ result.getEffectiveConfigs().forEach(configs::put);
+ }
+
+ ArrayNode errors = root.putArray("errors");
+ erroredChecks.forEach(errors::add);
+
+ try {
+ return JSON_MAPPER.writeValueAsString(root);
+ } catch (Exception e) {
+ throw new IllegalStateException("Failed to render health report as JSON", e);
+ }
+ }
+
+ public enum OutputFormat {
+ TABLE, JSON
+ }
+
+ public static class Config implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ @Parameter(names = {"--base-path", "-sp"}, description = "Base path of the Hudi table to check", required = true)
+ public String basePath = null;
+
+ @Parameter(names = {"--checks"}, description = "Comma-separated checks to run, or 'all'. Available: "
+ + "compaction, cleaner, savepoint, archival")
+ public String checks = "all";
+
+ @Parameter(names = {"--apply-all-defaults"}, description = "Evaluate checks against Hudi default values for any "
+ + "writer property not supplied via --props or --hoodie-conf. Off by default: without the writer properties "
+ + "the table was actually written with, a verdict derived from defaults can report a healthy table as "
+ + "unhealthy, so such checks are skipped instead.")
+ public Boolean applyAllDefaults = false;
+
+ @Parameter(names = {"--output"}, description = "Output format: TABLE or JSON")
+ public OutputFormat output = OutputFormat.TABLE;
+
+ @Parameter(names = {"--verbose", "-v"}, description = "Include the effective configuration each check used")
+ public Boolean verbose = false;
+
+ @Parameter(names = {"--props"}, description = "Path to a properties file on local fs or dfs, holding the writer "
+ + "configuration this table is written with")
+ public String propsFilePath = null;
+
+ @Parameter(names = {"--hoodie-conf"}, description = "Any writer configuration that can be set in the properties "
+ + "file (using --props) can also be passed on the command line here. This can be repeated",
+ splitter = IdentitySplitter.class)
+ public List<String> configs = new ArrayList<>();
+
+ @Parameter(names = {"--spark-master", "-ms"}, description = "Spark master for the checks that need one. "
+ + "Ignored when the job already has a master, as under spark-submit")
+ public String sparkMaster = "local[*]";
+
+ @Parameter(names = {"--help", "-h"}, help = true)
+ public Boolean help = false;
+
+ @Override
+ public String toString() {
+ return "HoodieTableHealthCheckerConfig {\n"
+ + " --base-path " + basePath + ", \n"
+ + " --checks " + checks + ", \n"
+ + " --apply-all-defaults " + applyAllDefaults + ", \n"
+ + " --spark-master " + sparkMaster + ", \n"
+ + " --output " + output + ", \n"
+ + " --verbose " + verbose + ", \n"
+ + " --props " + propsFilePath + ", \n"
+ + " --hoodie-conf " + configs
+ + "\n}";
+ }
+ }
+
+ public static void main(String[] args) {
+ Config cfg = new Config();
+ JCommander cmd = new JCommander(cfg, null, args);
+
+ if (cfg.help || args.length == 0) {
+ cmd.usage();
+ printAvailableChecks();
+ System.exit(EXIT_ERROR);
+ }
+
+ // Built unconditionally: a local context costs little, and building it only for the checks
+ // that need one would mean the set of selected checks decides whether the tool can run at all.
+ JavaSparkContext jsc = null;
+ try {
+ jsc = UtilHelpers.buildSparkContext("table-health-checker", cfg.sparkMaster, false);
+ System.exit(new HoodieTableHealthChecker(new Configuration(), cfg, Option.of(jsc)).run());
+ } catch (IllegalArgumentException e) {
+ System.err.println("ERROR: " + e.getMessage());
+ System.exit(EXIT_ERROR);
+ } catch (Throwable t) {
+ log.error("Failed to run table health check for {}", cfg, t);
+ System.exit(EXIT_ERROR);
+ } finally {
+ if (jsc != null) {
+ jsc.stop();
+ }
+ }
+ }
+
+ private static void printAvailableChecks() {
+ System.out.println("\nAvailable checks:");
+ allChecks().forEach(c -> System.out.printf(" %-12s %s%n", c.getName(), c.getDescription()));
+ System.out.println("\nExit codes: 0 = healthy, 2 = error, 3 = unhealthy, 4 = a check could not run\n");
+ }
+}
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
new file mode 100644
index 000000000000..ea40c0691d49
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/README.md
@@ -0,0 +1,225 @@
+<!--
+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.
+-->
+
+# HoodieTableHealthChecker
+
+Reports whether a Hudi table's table services are keeping up with ingestion. Meant to be run on a
+schedule — daily is typical — against the tables you are responsible for, with the exit code wired
+into whatever alerts you.
+
+The tool is strictly read-only. It inspects timeline state and reports; it never schedules or runs
+a table service, and never writes to the table.
+
+## Quick start
+
+```bash
+spark-submit \
+ --master local \
+ --class org.apache.hudi.utilities.HoodieTableHealthChecker \
+ $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.12-1.3.0-SNAPSHOT.jar \
+ --base-path s3://bucket/path/to/table \
+ --props s3://bucket/path/to/writer.properties
+```
+
+Run it with `spark-submit` rather than a bare `java -cp`: the tool builds a Spark context for the
+checks that need one, and the bundle relies on the Hadoop, Jackson and slf4j classes that
+`spark-submit` puts on the classpath. `--spark-master` sets the master when the job does not
+already have one; under `spark-submit` the submitted master wins.
+
+Most checks read only the timeline and finish in seconds whatever the table size. Checks that
+compare metadata-table listings against storage are the exception -- they scan the table, so size
+the job accordingly rather than assuming the whole run is cheap.
+
+## Why writer properties are required
+
+Nearly every question this tool answers is relative to how the table is written. Whether fifty
+delta commits since the last compaction is fine or alarming depends on the configured trigger.
+Whether a thousand instants in the active timeline is fine depends on the configured retention.
+
+Hudi supplies a default for each of these, so the tool *could* always print an answer. But when
+your table is written with different settings, that answer is wrong — and wrong in the direction
+that destroys trust, reporting a perfectly healthy table as unhealthy.
+
+So by default, **a check whose governing property you did not supply reports `SKIPPED` and names
+the property**, rather than guessing. Supply properties with `--props` or `--hoodie-conf`.
+
+If you know the table runs on stock settings, or you just want a rough look, pass
+`--apply-all-defaults` to evaluate against Hudi defaults explicitly.
+
+Every check echoes the configuration it actually reasoned with (`--verbose`, or the
+`effectiveConfigs` field in JSON), so a verdict can always be read alongside the values that
+produced it.
+
+## Options
+
+| Flag | Default | Meaning |
+|---|---|---|
+| `--base-path`, `-sp` | *(required)* | Base path of the table to check |
+| `--checks` | `all` | Comma-separated check names, or `all` |
+| `--apply-all-defaults` | `false` | Evaluate against Hudi defaults for any property not supplied |
+| `--output` | `TABLE` | `TABLE` for humans, `JSON` for machines |
+| `--verbose`, `-v` | `false` | Include the effective configuration each check used |
+| `--props` | none | Properties file (local fs or dfs) with the table's writer configuration |
+| `--hoodie-conf` | none | Single `key=value` writer property; repeatable |
+| `--help`, `-h` | | Usage, plus the list of available checks |
+
+## Checks
+
+### `compaction`
+
+Whether compaction is keeping pace with delta commits on a Merge-on-Read table. Skipped entirely on
+Copy-on-Write.
+
+Computes how many delta commits the configured trigger strategy allows before compaction should be
+scheduled, then compares that against how many have accumulated since the last completed
+compaction. A slack factor of 2.0 is applied before calling the table unhealthy: scheduling and
+execution are separate concerns, the execution strategy may legitimately defer individual file
+slices, and async compaction routinely trails ingestion by a cycle or two. The goal is catching
+compaction that has *stopped*, not policing jitter.
+
+When unhealthy, the report distinguishes two cases: nothing scheduled at all (the scheduler is not
+running), versus plans scheduled but never completed (the compactor is running but failing).
+
+Governing properties: `hoodie.compact.inline.trigger.strategy`,
+`hoodie.compact.inline.max.delta.commits`, `hoodie.compact.inline.max.delta.seconds`.
+
+### `cleaner`
+
+Whether the cleaner is keeping pace with write commits.
+
+Counts completed write commits since the last completed clean and compares against the configured
+trigger, with two allowances: the trigger is multiplied by 2.0 for async cleaning running on its
+own cadence, and the result floors at 5 commits. The floor matters because the trigger defaults to
+one commit -- without it, a cleaner two commits behind (routine under async cleaning, or when a
+clean briefly loses the table lock to a concurrent writer) would be an alert.
+
+When unhealthy, distinguishes a clean that is scheduled but stuck from no clean being scheduled at
+all. The last clean's retention watermark and files-deleted count are echoed for context.
+
+Governing property: `hoodie.clean.trigger.max.commits`. `hoodie.clean.policy` and
+`hoodie.clean.commits.retained` are echoed when supplied.
+
+### `savepoint`
+
+Whether a savepoint is holding back archival.
+
+Archival stops at the earliest savepoint and will not advance past it. A savepoint taken for a
+one-off restore and then forgotten therefore pins the active timeline indefinitely — the timeline
+grows without bound, every timeline load gets slower, and data files that cleaning would otherwise
+release stay alive. Nothing errors while this happens.
+
+Two signals are reported. A savepoint sitting further back than the configured archival retention
+is blocking archival right now. A savepoint older than 7 days is flagged separately as likely
+forgotten, since savepoints are normally short-lived.
+
+The remedy is always to release the savepoint once the operation it was taken for is complete
+(`savepoint delete` in `hudi-cli`).
+
+Governing property: `hoodie.keep.max.commits`.
+
+### `archival`
+
+Whether archival is keeping the active timeline bounded.
+
+Two independent signals. The first compares the active timeline against the configured retention,
+with 1.1x slack, since archival runs periodically rather than on every commit. The second is an
+absolute ceiling of 5000 completed instants — past roughly that point, loading and parsing the
+timeline is itself a per-operation performance problem regardless of what the configuration
+permits.
+
+This check does not attribute the cause. A stalled archival is most often a savepoint, which the
+`savepoint` check diagnoses specifically; read the two together.
+
+Governing property: `hoodie.keep.max.commits`.
+
+## Exit codes
+
+| Code | Meaning |
+|---|---|
+| `0` | Every check that ran reported healthy (or was skipped) |
+| `2` | Bad arguments, or the table could not be read |
+| `3` | At least one check reported unhealthy |
+| `4` | No check reported unhealthy, but at least one could not run |
+
+`1` is deliberately unused. The JVM returns it when the process fails to start at all — a missing
+class, a bad classpath — and a verdict sharing that code would make a setup mistake look like a
+finding about the table.
+
+Note that a run where everything skipped exits `0`. A skipped check is *silent* about health, not
+an assertion of it — if you want skips to be loud, check for `SKIPPED` in the output or use JSON.
+
+## JSON output
+
+```json
+{
+ "basePath" : "/path/to/table",
+ "tableType" : "MERGE_ON_READ",
+ "overallStatus" : "UNHEALTHY",
+ "checks" : [ {
+ "name" : "compaction",
+ "status" : "UNHEALTHY",
+ "summary" : "60 delta commit(s) have accumulated since compaction at 00001, past the threshold of 10",
+ "findings" : [
+ "Trigger strategy NUM_COMMITS expects compaction roughly every 5 delta commit(s); 60 have accumulated.",
+ "No compaction is scheduled. Check that compaction is enabled for this table and that the scheduling job is running."
+ ],
+ "effectiveConfigs" : {
+ "hoodie.compact.inline.trigger.strategy" : "NUM_COMMITS",
+ "hoodie.compact.inline.max.delta.commits" : "5",
+ "effective.unhealthy.threshold.delta.commits" : "10",
+ "observed.delta.commits.since.last.compaction" : "60"
+ }
+ } ],
+ "errors" : [ ]
+}
+```
+
+Keys prefixed `observed.` are measurements taken from the table; keys prefixed `effective.` are
+thresholds derived from the configuration. Everything else is a writer property as supplied.
+
+## Scheduling it
+
+```bash
+#!/usr/bin/env bash
+# Daily table health check.
+set -uo pipefail
+
+spark-submit --master local \
+ --class org.apache.hudi.utilities.HoodieTableHealthChecker "$HUDI_BUNDLE" \
+ --base-path "$TABLE" \
+ --props "$WRITER_PROPS" \
+ --output JSON > /tmp/health.json
+status=$?
+
+if [ $status -eq 3 ]; then
+ alert "Hudi table $TABLE is unhealthy" < /tmp/health.json
+elif [ $status -eq 4 ]; then
+ alert "Hudi health check: a check could not run for $TABLE" < /tmp/health.json
+elif [ $status -ne 0 ]; then
+ alert "Hudi health check failed to run for $TABLE (exit $status)" < /tmp/health.json
+fi
+```
+
+## Caveats
+
+- Verdicts are only as good as the writer properties you supply. Point `--props` at the same
+ properties your writer actually uses.
+- Thresholds are deliberately generous. This tool is built to catch a service that has stopped or
+ fallen badly behind, not to tune one that is running.
+- The table-level verdict is the most severe individual verdict. One unhealthy check makes the
+ table unhealthy, regardless of how many others passed.
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/ArchivalCadenceCheck.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/ArchivalCadenceCheck.java
new file mode 100644
index 000000000000..3ad074d2d62d
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/ArchivalCadenceCheck.java
@@ -0,0 +1,260 @@
+/*
+ * 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.model.HoodieCleaningPolicy;
+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.util.CleanerUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieArchivalConfig;
+import org.apache.hudi.config.HoodieCleanConfig;
+
+import java.text.ParseException;
+import java.time.Instant;
+
+import static org.apache.hudi.common.table.timeline.TimelineUtils.parseDateFromInstantTime;
+
+/**
+ * Checks whether archival is keeping the active timeline bounded.
+ *
+ * <p>Archival moves old instants out of the active timeline into the archived timeline. When it
+ * stops, the active timeline grows without limit, and because it is read and parsed on essentially
+ * every table operation, everything touching the table gets progressively slower. Unlike a failed
+ * write this produces no error -- the table simply degrades.
+ *
+ * <p>The configured {@code hoodie.keep.max.commits} is not on its own the bound the archiver
+ * enforces. {@code ArchivalUtils.getMinAndMaxInstantsToKeep} raises both bounds whenever the
+ * cleaner is retaining more history than archival would: archival must not move an instant off the
+ * active timeline while the cleaner still considers its files live, or incremental readers can be
+ * left reading files whose commits are gone. So a table with {@code clean.commits.retained=100} and
+ * {@code keep.max.commits=30} legitimately holds around a hundred instants, and calling that
+ * unhealthy is a false alarm. This check reproduces that calculation from the cleaner
+ * configuration, and reports the effective bound it derived.
+ *
+ * <p>Two independent signals are reported. The first compares the active timeline against that
+ * effective retention, with slack: archival runs periodically rather than on every commit, so a
+ * timeline slightly over the maximum is normal. The second is an absolute ceiling, because past a
+ * few thousand instants timeline loading is a problem regardless of what the configuration permits
+ * -- a table configured to retain a very large number of commits is misconfigured rather than
+ * healthy.
+ *
+ * <p>Attribution is deliberately limited. Archival also holds back for the earliest pending write
+ * instant, for compaction retention on MOR tables, and for the metadata table's own compaction, and
+ * this check cannot see which of those is binding. Where a pending instant is visible on the
+ * timeline it is named, because that is the one an operator can act on; otherwise the finding says
+ * what to look at rather than guessing. A blocked archival is most often a savepoint, which
+ * {@link SavepointArchivalBlockCheck} diagnoses specifically; the two are meant to be read
+ * together.
+ */
+public class ArchivalCadenceCheck implements HealthCheck {
+
+ static final String NAME = "archival";
+
+ /**
+ * How far past the configured retention the active timeline may grow before being called
+ * unhealthy. Archival is periodic, so the timeline routinely sits somewhat above the maximum
+ * between runs.
+ */
+ static final double RETENTION_SLACK_FACTOR = 1.1;
+
+ /**
+ * Active timeline size past which the table is called unhealthy regardless of configuration.
+ * Beyond roughly this many instants, loading and parsing the timeline is itself a performance
+ * problem on every operation against the table.
+ */
+ static final int MAX_REASONABLE_ACTIVE_INSTANTS = 5000;
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Whether archival is keeping the active timeline bounded";
+ }
+
+ @Override
+ public HealthCheckResult check(HealthCheckContext context) {
+ Option<String> missing = context.firstMissingConfig(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key());
+ if (missing.isPresent()) {
+ return context.skipForMissingConfig(NAME, missing.get());
+ }
+
+ int maxCommitsToKeep = Integer.parseInt(context.getProps().getString(
+ HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(),
+ HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.defaultValue()));
+ int minCommitsToKeep = Integer.parseInt(context.getProps().getString(
+ HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(),
+ HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.defaultValue()));
+
+ HoodieActiveTimeline activeTimeline = context.getMetaClient().getActiveTimeline();
+ // The archiver reasons over completed commits; getWriteTimeline() also counts requested and
+ // inflight compaction and clustering, which archival has not had the chance to consider.
+ HoodieTimeline completedCommitsTimeline =
+ activeTimeline.getCommitsTimeline().filterCompletedInstants();
+ int completedCommits = completedCommitsTimeline.countInstants();
+ int completedInstants = activeTimeline.filterCompletedInstants().countInstants();
+
+ int effectiveMaxCommitsToKeep =
+ effectiveMaxCommitsToKeep(context, completedCommitsTimeline, minCommitsToKeep, maxCommitsToKeep);
+ int retentionThreshold = (int) Math.ceil(RETENTION_SLACK_FACTOR * effectiveMaxCommitsToKeep);
+
+ HealthCheckResult.Builder result = HealthCheckResult.builder(NAME)
+ .config(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), maxCommitsToKeep)
+ .config(HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(), minCommitsToKeep)
+ .config("effective.max.commits.to.keep", effectiveMaxCommitsToKeep)
+ .config("effective.unhealthy.threshold.completed.commits", retentionThreshold)
+ .config("effective.absolute.ceiling.completed.instants", MAX_REASONABLE_ACTIVE_INSTANTS)
+ .config("observed.completed.commits", completedCommits)
+ .config("observed.completed.instants", completedInstants);
+
+ boolean overAbsoluteCeiling = completedInstants >= MAX_REASONABLE_ACTIVE_INSTANTS;
+ boolean overConfiguredRetention = completedCommits >= retentionThreshold;
+
+ if (!overAbsoluteCeiling && !overConfiguredRetention) {
+ return result.status(HealthStatus.HEALTHY)
+ .summary(String.format("Active timeline holds %d completed commit(s), within the "
+ + "effective threshold of %d", completedCommits, retentionThreshold))
+ .build();
+ }
+
+ if (overAbsoluteCeiling) {
+ result.finding(String.format(
+ "The active timeline holds %d completed instants. Past roughly %d, loading and parsing "
+ + "the timeline slows down every operation against the table, whatever the configured "
+ + "retention allows.",
+ completedInstants, MAX_REASONABLE_ACTIVE_INSTANTS));
+ }
+ if (overConfiguredRetention) {
+ if (effectiveMaxCommitsToKeep > maxCommitsToKeep) {
+ result.finding(String.format(
+ "The active timeline holds %d completed commits. The configured maximum is %d, but the "
+ + "cleaner's retention raises the bound the archiver enforces to %d, and the "
+ + "timeline is past even that.",
+ completedCommits, maxCommitsToKeep, effectiveMaxCommitsToKeep));
+ } else {
+ result.finding(String.format(
+ "The active timeline holds %d completed commits against a configured maximum of %d. "
+ + "Archival is not keeping up with ingestion.",
+ completedCommits, maxCommitsToKeep));
+ }
+ }
+
+ // Archival never moves past the earliest pending write instant, so when one is present it is
+ // the concrete thing to look at before anything else.
+ Option<HoodieInstant> earliestPending =
+ activeTimeline.getWriteTimeline().filterInflightsAndRequested().firstInstant();
+ if (earliestPending.isPresent()) {
+ result.finding(String.format(
+ "Archival cannot move past the earliest pending write instant, and %s (%s) is still %s. "
+ + "Complete or roll it back and archival will resume.",
+ earliestPending.get().requestedTime(), earliestPending.get().getAction(),
+ earliestPending.get().getState()));
+ }
+
+ return result.status(HealthStatus.UNHEALTHY)
+ .summary(String.format("Active timeline is not being archived down: %d completed commit(s), "
+ + "%d completed instant(s)", completedCommits, completedInstants))
+ .finding("Check the '" + SavepointArchivalBlockCheck.NAME + "' check first -- a savepoint is "
+ + "the most common reason archival stalls. Archival also holds back for compaction "
+ + "retention on MOR tables and for the metadata table's own compaction; if no savepoint "
+ + "and no pending instant explains this, confirm the archival service is running and "
+ + "look for repeated archival failures in the writer logs.")
+ .build();
+ }
+
+ /**
+ * The maximum the archiver will actually enforce, which is not always the configured one.
+ *
+ * <p>Mirrors {@code ArchivalUtils.getMinAndMaxInstantsToKeep}: when the cleaner retains more
+ * history than archival would, archival's bounds are raised to cover it, preserving the
+ * configured gap between min and max. Reproduced here rather than called because that method
+ * needs a {@code HoodieTable}, and this tool reads only the timeline.
+ *
+ * <p>Falls back to the configured maximum when the cleaner's earliest commit to retain cannot be
+ * determined, which keeps this check no worse than it was before.
+ */
+ private int effectiveMaxCommitsToKeep(HealthCheckContext context,
+ HoodieTimeline completedCommitsTimeline,
+ int minCommitsToKeep,
+ int maxCommitsToKeep) {
+ Option<HoodieInstant> earliestCommitToRetain =
+ earliestCommitToRetain(context, completedCommitsTimeline);
+ if (!earliestCommitToRetain.isPresent()) {
+ return maxCommitsToKeep;
+ }
+ int minBasedOnCleaning = completedCommitsTimeline
+ .findInstantsAfter(earliestCommitToRetain.get().requestedTime()).countInstants() + 2;
+ if (minCommitsToKeep >= minBasedOnCleaning) {
+ return maxCommitsToKeep;
+ }
+ return minBasedOnCleaning + maxCommitsToKeep - minCommitsToKeep;
+ }
+
+ private Option<HoodieInstant> earliestCommitToRetain(HealthCheckContext context,
+ HoodieTimeline completedCommitsTimeline) {
+ HoodieCleaningPolicy cleanerPolicy;
+ try {
+ cleanerPolicy = HoodieCleaningPolicy.valueOf(
+ context.getProps().getString(HoodieCleanConfig.CLEANER_POLICY.key(),
+ HoodieCleanConfig.CLEANER_POLICY.defaultValue()));
+ } catch (IllegalArgumentException e) {
+ return Option.empty();
+ }
+ int commitsRetained = Integer.parseInt(
+ context.getProps().getString(HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key(),
+ HoodieCleanConfig.CLEANER_COMMITS_RETAINED.defaultValue()));
+ int hoursRetained = Integer.parseInt(
+ context.getProps().getString(HoodieCleanConfig.CLEANER_HOURS_RETAINED.key(),
+ HoodieCleanConfig.CLEANER_HOURS_RETAINED.defaultValue()));
+ // Only KEEP_LATEST_BY_HOURS needs the latest commit as a wall-clock instant. Parsing it
+ // unconditionally would let an instant time this tool cannot parse disable the whole
+ // calculation, silently falling back to the configured bound.
+ Instant latestInstant = Instant.now();
+ if (HoodieCleaningPolicy.KEEP_LATEST_BY_HOURS == cleanerPolicy) {
+ Option<HoodieInstant> latestCommit = completedCommitsTimeline.lastInstant();
+ if (latestCommit.isPresent()) {
+ try {
+ latestInstant = parseDateFromInstantTime(latestCommit.get().requestedTime()).toInstant();
+ } catch (ParseException e) {
+ // Without a parsable latest commit the hours-based window cannot be placed; fall back to
+ // the configured bound rather than inventing one.
+ return Option.empty();
+ }
+ }
+ }
+ try {
+ return CleanerUtils.getEarliestCommitToRetain(
+ context.getMetaClient().getActiveTimeline().getCommitsTimeline(),
+ cleanerPolicy,
+ commitsRetained,
+ latestInstant,
+ hoursRetained,
+ context.getMetaClient().getTableConfig().getTimelineTimezone(),
+ Option.empty(),
+ Long.MAX_VALUE);
+ } catch (IllegalArgumentException e) {
+ // A cleaner policy we do not recognise.
+ return Option.empty();
+ }
+ }
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/CleanerCadenceCheck.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/CleanerCadenceCheck.java
new file mode 100644
index 000000000000..65d6e7dc7694
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/CleanerCadenceCheck.java
@@ -0,0 +1,245 @@
+/*
+ * 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.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.common.model.HoodieCleaningPolicy;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.WriteOperationType;
+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.CleanerUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieArchivalConfig;
+import org.apache.hudi.config.HoodieCleanConfig;
+
+/**
+ * Checks whether the cleaner is keeping pace with ingestion.
+ *
+ * <p>Cleaning is what releases storage: every commit leaves behind older file versions that stay on
+ * disk until a clean removes them. When the cleaner stops, the table keeps working and nothing
+ * errors -- storage simply grows, and with it the cost of every listing. Like the other cadence
+ * checks this one asks not "has a clean run recently" but whether write commits have accumulated
+ * well past the point where the configured trigger should have scheduled one.
+ *
+ * <p>Two allowances are made before calling the table unhealthy. The configured trigger is
+ * multiplied by a slack factor, since async cleaning trails ingestion by design. And a floor is
+ * applied on top, because the trigger defaults to one commit: without a floor, a cleaner that is
+ * two commits behind -- routine when cleaning is asynchronous or briefly loses a lock to a
+ * concurrent writer -- would be reported as a failure.
+ */
+public class CleanerCadenceCheck implements HealthCheck {
+
+ static final String NAME = "cleaner";
+
+ /** Multiplier on the configured trigger, absorbing async cleaning running on its own cadence. */
+ static final double TRIGGER_SLACK_FACTOR = 2.0;
+
+ /**
+ * Minimum number of commits since the last clean before the table can be called unhealthy,
+ * regardless of how low the trigger is set. Keeps a trigger of one commit from turning ordinary
+ * async lag into an alert.
+ */
+ static final int MIN_UNHEALTHY_COMMITS = 5;
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Whether the cleaner is keeping pace with write commits";
+ }
+
+ @Override
+ public HealthCheckResult check(HealthCheckContext context) throws Exception {
+ Option<String> missing = context.firstMissingConfig(HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key());
+ if (missing.isPresent()) {
+ return context.skipForMissingConfig(NAME, missing.get());
+ }
+
+ int triggerCommits = Integer.parseInt(context.getProps().getString(
+ HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key(),
+ HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.defaultValue()));
+ int threshold = Math.max((int) Math.ceil(TRIGGER_SLACK_FACTOR * triggerCommits), MIN_UNHEALTHY_COMMITS);
+
+ HoodieActiveTimeline activeTimeline = context.getMetaClient().getActiveTimeline();
+ HoodieTimeline completedWrites = activeTimeline.getCommitsTimeline().filterCompletedInstants();
+ Option<HoodieInstant> lastClean = activeTimeline.getCleanerTimeline().filterCompletedInstants().lastInstant();
+
+ HealthCheckResult.Builder result = HealthCheckResult.builder(NAME)
+ .config(HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key(), triggerCommits)
+ .config("effective.unhealthy.threshold.commits", threshold)
+ .config("observed.completed.write.commits", completedWrites.countInstants());
+ echoIfSupplied(context, result, HoodieCleanConfig.CLEANER_POLICY.key());
+ echoIfSupplied(context, result, HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key());
+
+ // A pure-append table never produces a clean instant, because nothing is ever
+ // superseded: every file group has a single slice and nothing is replaced. The
+ // cleaner is behaving correctly by doing nothing, so judging it on elapsed
+ // commits would flag such a table every day with no action available.
+ if (!lastClean.isPresent() && isAppendOnly(completedWrites)) {
+ return result.status(HealthStatus.HEALTHY)
+ .summary(String.format("No clean has run, and none is expected: all %d write commit(s) are "
+ + "inserts, so no file slice has been superseded", completedWrites.countInstants()))
+ .finding("Every completed write on this timeline is an insert or bulk_insert. Cleaning "
+ + "retires superseded file slices, and an append-only workload never creates any, so "
+ + "an empty cleaner timeline is the correct state rather than a stalled service. This "
+ + "check becomes meaningful once the table takes updates or deletes.")
+ .build();
+ }
+
+ int commitsSinceClean;
+ if (lastClean.isPresent()) {
+ commitsSinceClean = completedWrites.findInstantsAfter(lastClean.get().requestedTime()).countInstants();
+ result.config("observed.last.clean.instant", lastClean.get().requestedTime());
+ describeLastClean(context, lastClean.get(), result);
+ } else {
+ commitsSinceClean = completedWrites.countInstants();
+ result.config("observed.last.clean.instant", "none");
+ }
+ result.config("observed.commits.since.last.clean", commitsSinceClean);
+
+ if (commitsSinceClean <= threshold) {
+ return result.status(HealthStatus.HEALTHY)
+ .summary(lastClean.isPresent()
+ ? String.format("%d commit(s) since clean at %s, within the threshold of %d",
+ commitsSinceClean, lastClean.get().requestedTime(), threshold)
+ : String.format("%d commit(s) and no clean yet, within the threshold of %d",
+ commitsSinceClean, threshold))
+ .build();
+ }
+
+ result.status(HealthStatus.UNHEALTHY)
+ .summary(lastClean.isPresent()
+ ? String.format("%d commit(s) have accumulated since clean at %s, past the threshold of %d",
+ commitsSinceClean, lastClean.get().requestedTime(), threshold)
+ : String.format("%d commit(s) have accumulated and the table has never been cleaned, past the threshold of %d",
+ commitsSinceClean, threshold))
+ .finding(String.format(
+ "The cleaner is configured to run every %d commit(s); %d have landed without one completing.",
+ triggerCommits, commitsSinceClean));
+
+ if (activeTimeline.getCleanerTimeline().filterInflightsAndRequested().countInstants() > 0) {
+ result.finding("A clean is scheduled or in flight but has not completed. Look for a failing or "
+ + "stuck clean in the writer logs; a clean that repeatedly fails to acquire the table lock "
+ + "under concurrent writers is a common cause.");
+ } else {
+ // Name the configuration conflict before blaming the service: if archival trims the timeline
+ // below what the cleaner needs to see, the cleaner is starved by construction and no amount
+ // of checking that it is enabled will help.
+ Option<String> starvedByArchival = archivalStarvesCleaner(context);
+ if (starvedByArchival.isPresent()) {
+ result.finding(starvedByArchival.get());
+ } else {
+ result.finding("No clean is scheduled. Check that cleaning is enabled for this table ("
+ + HoodieCleanConfig.AUTO_CLEAN.key() + ") and that the job responsible for it is running.");
+ }
+ }
+ return result.build();
+ }
+
+ /**
+ * Whether archival is configured to trim the timeline before the cleaner can act on it.
+ *
+ * <p>Under {@code KEEP_LATEST_COMMITS} the cleaner only retains history once it can see more than
+ * {@code hoodie.clean.commits.retained} completed commits. If {@code hoodie.keep.min.commits} is
+ * at or below that number, archival removes instants from the active timeline first and the
+ * cleaner never reaches its trigger -- file versions accumulate while no clean is ever scheduled.
+ * Hudi warns about this at write time and the archiver raises its own bounds to compensate, but a
+ * table already in this state reads as "cleaning is not running" when the cause is the pairing of
+ * the two configs.
+ *
+ * @return the finding to report, or empty when the configuration is not the explanation.
+ */
+ private static Option<String> archivalStarvesCleaner(HealthCheckContext context) {
+ String policy = context.getProps().getString(HoodieCleanConfig.CLEANER_POLICY.key(),
+ HoodieCleanConfig.CLEANER_POLICY.defaultValue());
+ if (!HoodieCleaningPolicy.KEEP_LATEST_COMMITS.name().equals(policy)) {
+ return Option.empty();
+ }
+ int commitsRetained = Integer.parseInt(
+ context.getProps().getString(HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key(),
+ HoodieCleanConfig.CLEANER_COMMITS_RETAINED.defaultValue()));
+ int minCommitsToKeep = Integer.parseInt(
+ context.getProps().getString(HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(),
+ HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.defaultValue()));
+ if (minCommitsToKeep > commitsRetained) {
+ return Option.empty();
+ }
+ return Option.of(String.format(
+ "%s=%d is not greater than %s=%d, so archival trims the active timeline before the cleaner "
+ + "has enough commits to act on. Raise %s above %d. Until then file versions accumulate "
+ + "and no clean is scheduled, whatever %s is set to.",
+ HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(), minCommitsToKeep,
+ HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key(), commitsRetained,
+ HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(), commitsRetained,
+ HoodieCleanConfig.AUTO_CLEAN.key()));
+ }
+
+ /**
+ * Whether every completed write on the timeline is an insert.
+ *
+ * <p>Read from the commit metadata rather than from
+ * {@code hoodie.datasource.write.operation}, because what the table has actually received is the
+ * question -- a writer configured for upsert that has only ever inserted still has nothing to
+ * clean, and a property file can be stale or absent.
+ *
+ * <p>Conservative on failure: an unreadable or unknown commit means this returns false and the
+ * normal cadence judgement applies. Suppressing a real finding is the worse error here.
+ */
+ private static boolean isAppendOnly(HoodieTimeline completedWrites) {
+ if (completedWrites.empty()) {
+ return false;
+ }
+ for (HoodieInstant instant : completedWrites.getInstants()) {
+ try {
+ HoodieCommitMetadata metadata = TimelineUtils.getCommitMetadata(instant, completedWrites);
+ if (!WriteOperationType.isInsertWithoutReplace(metadata.getOperationType())) {
+ return false;
+ }
+ } catch (Exception e) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /** Surfaces what the last clean retained, so the operator can see the retention watermark moving. */
+ private static void describeLastClean(HealthCheckContext context, HoodieInstant lastClean,
+ HealthCheckResult.Builder result) {
+ try {
+ HoodieCleanMetadata metadata = CleanerUtils.getCleanerMetadata(context.getMetaClient(), lastClean);
+ result.config("observed.last.clean.earliest.commit.retained", metadata.getEarliestCommitToRetain());
+ result.config("observed.last.clean.files.deleted", metadata.getTotalFilesDeleted());
+ } catch (Exception e) {
+ // The metadata is informational only; an unreadable clean must not fail the verdict.
+ result.config("observed.last.clean.metadata", "unreadable: " + e.getMessage());
+ }
+ }
+
+ private static void echoIfSupplied(HealthCheckContext context, HealthCheckResult.Builder result, String key) {
+ if (context.getProps().containsKey(key)) {
+ result.config(key, context.getProps().getString(key));
+ }
+ }
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/CompactionCadenceCheck.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/CompactionCadenceCheck.java
new file mode 100644
index 000000000000..a8099e6e25b2
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/CompactionCadenceCheck.java
@@ -0,0 +1,206 @@
+/*
+ * 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.model.HoodieTableType;
+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.config.HoodieCompactionConfig;
+import org.apache.hudi.table.action.compact.CompactionTriggerStrategy;
+
+import java.sql.Date;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+
+/**
+ * Checks whether compaction is keeping pace with ingestion on a Merge-on-Read table.
+ *
+ * <p>The question is not "has compaction run recently" -- a quiet table legitimately has no recent
+ * compaction. It is whether delta commits have accumulated well past the point where the configured
+ * trigger should have fired. So the check computes how many delta commits the writer's trigger
+ * strategy says should elapse before compaction is scheduled, then compares that against how many
+ * have actually accumulated since the last completed compaction.
+ *
+ * <p>A slack factor is applied before calling the table unhealthy. Compaction scheduling and
+ * compaction execution are separate: a plan can be scheduled promptly while the execution strategy
+ * defers individual file slices, and async compaction can trail ingestion by a cycle or two without
+ * anything being wrong. Alerting at exactly the trigger threshold would fire on every healthy MOR
+ * table. The factor below is deliberately generous -- this check is meant to catch compaction that
+ * has stopped or fallen badly behind, not to police scheduling jitter.
+ */
+public class CompactionCadenceCheck implements HealthCheck {
+
+ static final String NAME = "compaction";
+
+ /**
+ * How far past the configured trigger point delta commits may accumulate before the table is
+ * called unhealthy. Absorbs the gap between scheduling and execution, and async table services
+ * running on their own cadence.
+ */
+ static final double TRIGGER_SLACK_FACTOR = 2.0;
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Whether compaction is keeping pace with delta commits on a Merge-on-Read table";
+ }
+
+ @Override
+ public boolean appliesTo(HealthCheckContext context) {
+ return context.getMetaClient().getTableType() == HoodieTableType.MERGE_ON_READ;
+ }
+
+ @Override
+ public HealthCheckResult check(HealthCheckContext context) {
+ if (!appliesTo(context)) {
+ return HealthCheckResult.skipped(NAME, "Table is Copy-on-Write, which has no compaction");
+ }
+
+ Option<String> missing = context.firstMissingConfig(
+ HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.key(),
+ HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key());
+ if (missing.isPresent()) {
+ return context.skipForMissingConfig(NAME, missing.get());
+ }
+
+ CompactionTriggerStrategy strategy = CompactionTriggerStrategy.valueOf(
+ context.getProps().getString(
+ HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.key(),
+ HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.defaultValue()));
+ int maxDeltaCommits = Integer.parseInt(context.getProps().getString(
+ HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(),
+ HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.defaultValue()));
+ int maxDeltaSeconds = Integer.parseInt(context.getProps().getString(
+ HoodieCompactionConfig.INLINE_COMPACT_TIME_DELTA_SECONDS.key(),
+ HoodieCompactionConfig.INLINE_COMPACT_TIME_DELTA_SECONDS.defaultValue()));
+
+ HoodieActiveTimeline activeTimeline = context.getMetaClient().getActiveTimeline();
+ int expectedBeforeTrigger = expectedDeltaCommitsBeforeTrigger(
+ strategy, maxDeltaCommits, maxDeltaSeconds, activeTimeline);
+ int threshold = (int) Math.ceil(TRIGGER_SLACK_FACTOR * expectedBeforeTrigger);
+
+ Option<Pair<HoodieTimeline, HoodieInstant>> sinceLastCompaction =
+ CompactionUtils.getCompletedDeltaCommitsSinceLatestCompaction(activeTimeline);
+
+ HealthCheckResult.Builder result = HealthCheckResult.builder(NAME)
+ .config(HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.key(), strategy)
+ .config(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(), maxDeltaCommits)
+ .config(HoodieCompactionConfig.INLINE_COMPACT_TIME_DELTA_SECONDS.key(), maxDeltaSeconds)
+ .config("effective.unhealthy.threshold.delta.commits", threshold);
+
+ if (!sinceLastCompaction.isPresent()) {
+ return result.status(HealthStatus.HEALTHY)
+ .summary("No delta commits have landed yet; nothing for compaction to fall behind on")
+ .build();
+ }
+
+ int pendingDeltaCommits = sinceLastCompaction.get().getLeft().countInstants();
+ int pendingCompactions = CompactionUtils.getPendingCompactionInstantTimes(context.getMetaClient()).size();
+ HoodieInstant lastCompaction = sinceLastCompaction.get().getRight();
+
+ result.config("observed.delta.commits.since.last.compaction", pendingDeltaCommits)
+ .config("observed.pending.compactions", pendingCompactions)
+ .config("observed.last.compaction.instant", lastCompaction.requestedTime());
+
+ if (pendingDeltaCommits <= threshold) {
+ return result.status(HealthStatus.HEALTHY)
+ .summary(String.format(
+ "%d delta commit(s) since compaction at %s, within the threshold of %d",
+ pendingDeltaCommits, lastCompaction.requestedTime(), threshold))
+ .build();
+ }
+
+ result.status(HealthStatus.UNHEALTHY)
+ .summary(String.format(
+ "%d delta commit(s) have accumulated since compaction at %s, past the threshold of %d",
+ pendingDeltaCommits, lastCompaction.requestedTime(), threshold))
+ .finding(String.format(
+ "Trigger strategy %s expects compaction roughly every %d delta commit(s); %d have accumulated.",
+ strategy, expectedBeforeTrigger, pendingDeltaCommits));
+
+ if (pendingCompactions > 0) {
+ result.finding(String.format(
+ "%d compaction(s) are already scheduled but not complete -- compaction is being scheduled "
+ + "and not executed. Check that an async compactor is running and not failing.",
+ pendingCompactions));
+ } else {
+ result.finding("No compaction is scheduled. Check that compaction is enabled for this table "
+ + "and that the scheduling job is running.");
+ }
+
+ return result.build();
+ }
+
+ /**
+ * Number of delta commits the configured trigger strategy allows before compaction should be
+ * scheduled. For the time-based strategies this is expressed in delta commits too, by counting
+ * how many have landed inside the configured time window, so that the caller has a single unit
+ * to compare against.
+ */
+ private int expectedDeltaCommitsBeforeTrigger(CompactionTriggerStrategy strategy,
+ int maxDeltaCommits,
+ int maxDeltaSeconds,
+ HoodieActiveTimeline activeTimeline) {
+ HoodieTimeline completedDeltaCommits = activeTimeline.getDeltaCommitTimeline().filterCompletedInstants();
+ String windowStart = TimelineUtils.formatDate(
+ Date.from(Instant.now().minus(maxDeltaSeconds, ChronoUnit.SECONDS)));
+ int deltaCommitsInTimeWindow = completedDeltaCommits.findInstantsAfter(windowStart).countInstants();
+
+ switch (strategy) {
+ case NUM_COMMITS:
+ return maxDeltaCommits;
+
+ case NUM_COMMITS_AFTER_LAST_REQUEST:
+ // The budget restarts from the last compaction *request*, so delta commits that landed
+ // between the last completed compaction and that request are already spoken for.
+ Option<HoodieInstant> lastCompleted = activeTimeline.getCommitTimeline().filterCompletedInstants().lastInstant();
+ Option<HoodieInstant> lastRequested = activeTimeline.filterPendingCompactionTimeline().lastInstant();
+ if (lastCompleted.isPresent() && lastRequested.isPresent()) {
+ int between = completedDeltaCommits
+ .findInstantsInRange(lastCompleted.get().requestedTime(), lastRequested.get().requestedTime())
+ .countInstants();
+ return between + maxDeltaCommits;
+ }
+ return maxDeltaCommits;
+
+ case TIME_ELAPSED:
+ return deltaCommitsInTimeWindow;
+
+ case NUM_OR_TIME:
+ // Either condition fires it, so the earlier one governs.
+ return Math.min(deltaCommitsInTimeWindow, maxDeltaCommits);
+
+ case NUM_AND_TIME:
+ // Both must hold, so the later one governs.
+ return Math.max(deltaCommitsInTimeWindow, maxDeltaCommits);
+
+ default:
+ throw new IllegalArgumentException("Unknown compaction trigger strategy: " + strategy);
+ }
+ }
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheck.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheck.java
new file mode 100644
index 000000000000..9b115e847f4b
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheck.java
@@ -0,0 +1,53 @@
+/*
+ * 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;
+
+/**
+ * One diagnosis of whether a table service is keeping up.
+ *
+ * <p>Implementations are read-only: they inspect timeline and file-system state and report, never
+ * scheduling or running a table service. Each is independently selectable from the command line via
+ * {@link #getName()}, so an operator can run the whole battery daily or a single check while
+ * investigating.
+ *
+ * <p>An implementation should not throw to signal an unhealthy table -- that is what
+ * {@link HealthStatus#UNHEALTHY} is for. Exceptions are reserved for a check that could not reach a
+ * verdict at all, and the runner records those separately so one broken check does not hide the
+ * results of the others.
+ */
+public interface HealthCheck {
+
+ /** Command-line name of this check, as accepted by {@code --checks}. Lowercase, hyphenated. */
+ String getName();
+
+ /** One line describing what this check looks at, shown in {@code --help}. */
+ String getDescription();
+
+ /**
+ * Whether this check applies to the given table at all, independent of configuration. A check
+ * that does not apply -- compaction on a Copy-on-Write table, for instance -- is reported as
+ * {@link HealthStatus#SKIPPED} rather than run.
+ */
+ default boolean appliesTo(HealthCheckContext context) {
+ return true;
+ }
+
+ /** Runs the check and returns its verdict. */
+ HealthCheckResult check(HealthCheckContext context) throws Exception;
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheckContext.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheckContext.java
new file mode 100644
index 000000000000..034acabeef50
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheckContext.java
@@ -0,0 +1,136 @@
+/*
+ * 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.TypedProperties;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+
+import org.apache.spark.api.java.JavaSparkContext;
+
+/**
+ * Everything a {@link HealthCheck} is given: the table, the writer properties it is judged against,
+ * and whether unset properties may fall back to Hudi defaults.
+ *
+ * <p>The distinction that {@code applyAllDefaults} draws is the point of this class. A health verdict
+ * is derived by comparing observed table state against what the writer configuration says should
+ * happen -- how many delta commits before compaction triggers, how many commits the cleaner retains,
+ * how many instants archival keeps. Hudi supplies a default for every one of those, so a checker can
+ * always produce an answer. But when the table is actually written with different settings, that
+ * answer is wrong, and wrong in the worst direction: a table that is perfectly healthy under its own
+ * configuration gets reported as UNHEALTHY, and an operator learns to distrust the tool.
+ *
+ * <p>So by default a check whose governing property was not supplied reports
+ * {@link HealthStatus#SKIPPED} rather than guessing. Operators who know their table runs on stock
+ * settings, or who just want a rough look, opt into defaults explicitly with
+ * {@code --apply-all-defaults}.
+ */
+public class HealthCheckContext {
+
+ private final HoodieTableMetaClient metaClient;
+ private final TypedProperties props;
+ private final boolean applyAllDefaults;
+ private final Option<JavaSparkContext> jsc;
+
+ public HealthCheckContext(HoodieTableMetaClient metaClient, TypedProperties props, boolean applyAllDefaults) {
+ this(metaClient, props, applyAllDefaults, Option.empty());
+ }
+
+ public HealthCheckContext(HoodieTableMetaClient metaClient, TypedProperties props, boolean applyAllDefaults,
+ Option<JavaSparkContext> jsc) {
+ this.metaClient = metaClient;
+ this.props = props;
+ this.applyAllDefaults = applyAllDefaults;
+ this.jsc = jsc;
+ }
+
+ public HoodieTableMetaClient getMetaClient() {
+ return metaClient;
+ }
+
+ /**
+ * The Spark context, when the driver was given one.
+ *
+ * <p>Checks that only read the timeline do not need it. Two kinds do: one delegating to a
+ * Spark-based tool, as the metadata-table sync check does with
+ * {@code HoodieMetadataTableValidator}, and one whose work does not fit on the driver -- walking
+ * file slices per file group for compaction, or file versions for cleaning, is a driver-side loop
+ * today and will not stay one on a table with hundreds of thousands of file groups.
+ *
+ * <p>A check that needs it must handle its absence by reporting {@link HealthStatus#SKIPPED} and
+ * saying what it needed, rather than failing the run: the checker is meant to stay usable
+ * without a cluster for the checks that do not require one.
+ */
+ public Option<JavaSparkContext> getJsc() {
+ return jsc;
+ }
+
+ public TypedProperties getProps() {
+ return props;
+ }
+
+ public boolean isApplyAllDefaults() {
+ return applyAllDefaults;
+ }
+
+ /**
+ * Whether every one of {@code keys} was supplied by the operator, or defaults were explicitly
+ * allowed. Checks call this before reading configuration so that a missing key becomes a
+ * {@link HealthStatus#SKIPPED} verdict instead of a verdict derived from a value the table may
+ * never have been written with.
+ */
+ public boolean hasConfigsOrDefaultsAllowed(String... keys) {
+ if (applyAllDefaults) {
+ return true;
+ }
+ for (String key : keys) {
+ if (!props.containsKey(key)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * The first of {@code keys} that the operator did not supply, for naming in a skip reason.
+ * Empty when all are present or defaults are allowed.
+ */
+ public Option<String> firstMissingConfig(String... keys) {
+ if (applyAllDefaults) {
+ return Option.empty();
+ }
+ for (String key : keys) {
+ if (!props.containsKey(key)) {
+ return Option.of(key);
+ }
+ }
+ return Option.empty();
+ }
+
+ /**
+ * A skip result naming the missing property and how to proceed. Phrased for an operator reading
+ * the report, who needs to know both what to supply and that there is a way to run without it.
+ */
+ public HealthCheckResult skipForMissingConfig(String checkName, String missingKey) {
+ return HealthCheckResult.skipped(checkName, String.format(
+ "Writer property '%s' was not supplied, so this check cannot judge the table against the "
+ + "settings it is actually written with. Pass it via --props or --hoodie-conf, or rerun "
+ + "with --apply-all-defaults to evaluate against Hudi defaults.", missingKey));
+ }
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheckResult.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheckResult.java
new file mode 100644
index 000000000000..7a6f1ac6488f
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheckResult.java
@@ -0,0 +1,117 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Verdict of one {@link HealthCheck} over one table.
+ *
+ * <p>Carries three things beyond the {@link HealthStatus}: a one-line {@code summary} for the
+ * operator, the {@code effectiveConfigs} the check actually reasoned with, and {@code findings}
+ * naming what is behind. Echoing the configuration back matters because a verdict derived from
+ * writer properties is only meaningful alongside the properties it used -- the same table is
+ * healthy or unhealthy depending on the retention and trigger settings it is written with.
+ */
+public class HealthCheckResult {
+
+ private final String checkName;
+ private final HealthStatus status;
+ private final String summary;
+ private final Map<String, String> effectiveConfigs;
+ private final List<String> findings;
+
+ private HealthCheckResult(String checkName, HealthStatus status, String summary,
+ Map<String, String> effectiveConfigs, List<String> findings) {
+ this.checkName = checkName;
+ this.status = status;
+ this.summary = summary;
+ this.effectiveConfigs = Collections.unmodifiableMap(new LinkedHashMap<>(effectiveConfigs));
+ this.findings = Collections.unmodifiableList(new ArrayList<>(findings));
+ }
+
+ public static Builder builder(String checkName) {
+ return new Builder(checkName);
+ }
+
+ /** A check that did not run, with the reason stated for the operator. */
+ public static HealthCheckResult skipped(String checkName, String reason) {
+ return builder(checkName).status(HealthStatus.SKIPPED).summary(reason).build();
+ }
+
+ public String getCheckName() {
+ return checkName;
+ }
+
+ public HealthStatus getStatus() {
+ return status;
+ }
+
+ public String getSummary() {
+ return summary;
+ }
+
+ public Map<String, String> getEffectiveConfigs() {
+ return effectiveConfigs;
+ }
+
+ public List<String> getFindings() {
+ return findings;
+ }
+
+ public static class Builder {
+ private final String checkName;
+ private final Map<String, String> effectiveConfigs = new LinkedHashMap<>();
+ private final List<String> findings = new ArrayList<>();
+ private HealthStatus status = HealthStatus.SKIPPED;
+ private String summary = "";
+
+ private Builder(String checkName) {
+ this.checkName = checkName;
+ }
+
+ public Builder status(HealthStatus status) {
+ this.status = status;
+ return this;
+ }
+
+ public Builder summary(String summary) {
+ this.summary = summary;
+ return this;
+ }
+
+ public Builder config(String key, Object value) {
+ this.effectiveConfigs.put(key, String.valueOf(value));
+ return this;
+ }
+
+ public Builder finding(String finding) {
+ this.findings.add(finding);
+ return this;
+ }
+
+ public HealthCheckResult build() {
+ return new HealthCheckResult(checkName, status, summary, effectiveConfigs, findings);
+ }
+ }
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthStatus.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthStatus.java
new file mode 100644
index 000000000000..c824d80a93e6
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthStatus.java
@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+/**
+ * Outcome of a single {@link HealthCheck}.
+ *
+ * <p>Ordered from least to most severe so that a table-level verdict can be derived by taking the
+ * maximum over the individual check outcomes; see {@link #max(HealthStatus, HealthStatus)}.
+ */
+public enum HealthStatus {
+ /**
+ * The check did not run. Either the governing writer configuration was not supplied, or the check
+ * does not apply to this table (for example, compaction on a Copy-on-Write table). Never a failure:
+ * a skipped check is silent about the table's health rather than asserting it is fine.
+ */
+ SKIPPED,
+
+ /** The check ran and the table service is keeping up. */
+ HEALTHY,
+
+ /** The check ran and found the table service falling behind. */
+ UNHEALTHY;
+
+ public static HealthStatus max(HealthStatus left, HealthStatus right) {
+ return left.compareTo(right) >= 0 ? left : right;
+ }
+
+ public boolean isUnhealthy() {
+ return this == UNHEALTHY;
+ }
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/SavepointArchivalBlockCheck.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/SavepointArchivalBlockCheck.java
new file mode 100644
index 000000000000..fdf32076fff4
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/health/SavepointArchivalBlockCheck.java
@@ -0,0 +1,150 @@
+/*
+ * 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.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.TimelineUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieArchivalConfig;
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Checks for savepoints that are holding back archival.
+ *
+ * <p>Archival stops at the earliest savepoint and will not advance past it. A savepoint taken for a
+ * one-off restore and then forgotten therefore pins the active timeline indefinitely. The timeline
+ * grows without bound, every timeline load gets slower, and because archival is blocked the
+ * cleaner's retention interacts with it in ways that keep data files alive too. This is a quiet
+ * failure: the table keeps accepting writes and nothing errors.
+ *
+ * <p>Two signals are reported. A savepoint sitting further back than the configured archival
+ * retention is actively blocking archival now. A savepoint older than {@link #STALE_SAVEPOINT_DAYS}
+ * is flagged separately as likely forgotten, since savepoints are normally short-lived.
+ *
+ * <p>The remedy this check points at is always to release the savepoint once the operation it was
+ * taken for is done. It deliberately does not suggest loosening archival to step over savepoints.
+ */
+public class SavepointArchivalBlockCheck implements HealthCheck {
+
+ static final String NAME = "savepoint";
+
+ /**
+ * Age past which a savepoint is called out as probably forgotten. Savepoints are normally taken
+ * around a specific operation and released soon after; one lingering for weeks is usually an
+ * oversight rather than an intent.
+ */
+ static final int STALE_SAVEPOINT_DAYS = 7;
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Whether a savepoint is blocking archival or has been left behind";
+ }
+
+ @Override
+ public HealthCheckResult check(HealthCheckContext context) {
+ Option<String> missing = context.firstMissingConfig(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key());
+ if (missing.isPresent()) {
+ return context.skipForMissingConfig(NAME, missing.get());
+ }
+
+ int maxCommitsToKeep = Integer.parseInt(context.getProps().getString(
+ HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(),
+ HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.defaultValue()));
+ HoodieTimeline writeTimeline = context.getMetaClient().getActiveTimeline().getWriteTimeline();
+ List<HoodieInstant> savepoints = context.getMetaClient().getActiveTimeline()
+ .getSavePointTimeline().filterCompletedInstants().getInstants();
+
+ HealthCheckResult.Builder result = HealthCheckResult.builder(NAME)
+ .config(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), maxCommitsToKeep)
+ .config("observed.savepoint.count", savepoints.size());
+
+ if (savepoints.isEmpty()) {
+ return result.status(HealthStatus.HEALTHY).summary("No savepoints on the table").build();
+ }
+
+ List<String> savepointTimes = savepoints.stream()
+ .map(HoodieInstant::requestedTime).collect(Collectors.toList());
+ result.config("observed.savepoints", String.join(",", savepointTimes));
+
+ List<String> blocking = savepointsBeyondRetention(writeTimeline, savepointTimes, maxCommitsToKeep);
+ List<String> stale = savepointTimes.stream()
+ .filter(SavepointArchivalBlockCheck::isOlderThanStaleThreshold)
+ .collect(Collectors.toList());
+
+ if (!stale.isEmpty()) {
+ result.finding(String.format(
+ "Savepoint(s) %s are more than %d days old. If the operation they were taken for is "
+ + "finished, delete them so archival can advance.",
+ String.join(", ", stale), STALE_SAVEPOINT_DAYS));
+ }
+
+ if (blocking.isEmpty()) {
+ return result.status(HealthStatus.HEALTHY)
+ .summary(String.format("%d savepoint(s) present, none older than the archival retention of %d commits",
+ savepoints.size(), maxCommitsToKeep))
+ .build();
+ }
+
+ return result.status(HealthStatus.UNHEALTHY)
+ .summary(String.format(
+ "%d savepoint(s) older than the archival retention are blocking archival: %s",
+ blocking.size(), String.join(", ", blocking)))
+ .finding(String.format(
+ "Archival stops at the earliest savepoint. The active timeline cannot shrink past %s "
+ + "until that savepoint is released.", blocking.get(0)))
+ .finding("Delete the savepoint once the operation it was taken for is complete, using the "
+ + "'savepoint delete' command in hudi-cli or SavepointsCommand.")
+ .build();
+ }
+
+ /**
+ * Savepoints that fall outside the window archival is allowed to retain -- that is, old enough
+ * that archival would have removed them by now were they not savepointed.
+ */
+ private static List<String> savepointsBeyondRetention(HoodieTimeline writeTimeline,
+ List<String> savepointTimes,
+ int maxCommitsToKeep) {
+ List<HoodieInstant> instants = writeTimeline.getInstants();
+ if (instants.size() <= maxCommitsToKeep) {
+ return Collections.emptyList();
+ }
+ return instants.subList(0, instants.size() - maxCommitsToKeep).stream()
+ .map(HoodieInstant::requestedTime)
+ .filter(savepointTimes::contains)
+ .collect(Collectors.toList());
+ }
+
+ private static boolean isOlderThanStaleThreshold(String instantTime) {
+ return TimelineUtils.parseDateFromInstantTimeSafely(instantTime)
+ .map(date -> date.toInstant()
+ .isBefore(Instant.now().minus(STALE_SAVEPOINT_DAYS, ChronoUnit.DAYS)))
+ .orElse(false);
+ }
+}
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
new file mode 100644
index 000000000000..2f194bc4e2c3
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTableHealthChecker.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;
+
+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.common.testutils.HoodieTestUtils;
+import org.apache.hudi.config.HoodieArchivalConfig;
+import org.apache.hudi.config.HoodieCleanConfig;
+import org.apache.hudi.config.HoodieCompactionConfig;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link HoodieTableHealthChecker} end to end: check selection, output rendering, and the
+ * exit codes a scheduled run depends on.
+ */
+public class TestHoodieTableHealthChecker {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ @TempDir
+ Path tempDir;
+
+ private String basePath;
+
+ @BeforeEach
+ public void setUp() {
+ basePath = tempDir.resolve("table").toString();
+ }
+
+ private HoodieTableHealthChecker.Config configFor(String checks) {
+ HoodieTableHealthChecker.Config cfg = new HoodieTableHealthChecker.Config();
+ cfg.basePath = basePath;
+ cfg.checks = checks;
+ return cfg;
+ }
+
+ /** Runs the checker, returning its exit code and whatever it printed. */
+ private Result runChecker(HoodieTableHealthChecker.Config cfg) {
+ PrintStream originalOut = System.out;
+ ByteArrayOutputStream captured = new ByteArrayOutputStream();
+ try {
+ System.setOut(new PrintStream(captured, true, StandardCharsets.UTF_8.name()));
+ int exitCode = new HoodieTableHealthChecker(new Configuration(), cfg).run();
+ return new Result(exitCode, new String(captured.toByteArray(), StandardCharsets.UTF_8));
+ } catch (RuntimeException e) {
+ // Surface the tool's own unchecked exceptions unwrapped, so tests can assert on their type.
+ throw e;
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ } finally {
+ System.setOut(originalOut);
+ }
+ }
+
+ private HoodieTableMetaClient healthyMorTable() throws Exception {
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.MERGE_ON_READ);
+ HoodieTestTable.of(metaClient).addDeltaCommit("001").addDeltaCommit("002");
+ return metaClient;
+ }
+
+ private HoodieTableMetaClient laggingMorTable() throws Exception {
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.MERGE_ON_READ);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = 1; i <= 60; i++) {
+ table.addDeltaCommit(String.format("%05d", i));
+ }
+ return metaClient;
+ }
+
+ @Test
+ public void healthyTableExitsZero() throws Exception {
+ // given a table whose services are keeping up, with writer properties supplied
+ healthyMorTable();
+ HoodieTableHealthChecker.Config cfg = configFor("all");
+ cfg.configs = Arrays.asList(
+ HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.key() + "=NUM_COMMITS",
+ HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key() + "=5",
+ HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key() + "=1",
+ HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key() + "=30");
+
+ // when the checker runs
+ Result result = runChecker(cfg);
+
+ // then it exits 0 and says so
+ assertEquals(HoodieTableHealthChecker.EXIT_HEALTHY, result.exitCode);
+ assertTrue(result.output.contains("overall : HEALTHY"), result.output);
+ }
+
+ @Test
+ public void unhealthyTableExitsOne() throws Exception {
+ // given a table badly behind on both compaction and archival
+ laggingMorTable();
+ HoodieTableHealthChecker.Config cfg = configFor("all");
+ cfg.configs = Arrays.asList(
+ HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.key() + "=NUM_COMMITS",
+ HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key() + "=5",
+ HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key() + "=1",
+ HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key() + "=10");
+
+ // when the checker runs
+ Result result = runChecker(cfg);
+
+ // then it exits 1, so a cron job notices without parsing anything
+ assertEquals(HoodieTableHealthChecker.EXIT_UNHEALTHY, result.exitCode);
+ assertTrue(result.output.contains("overall : UNHEALTHY"), result.output);
+ }
+
+ @Test
+ public void tableWithNoSuppliedPropertiesSkipsEverythingAndStillExitsZero() throws Exception {
+ // given a badly lagging table, but no writer properties to judge it against
+ laggingMorTable();
+
+ // when the checker runs with nothing supplied
+ Result result = runChecker(configFor("all"));
+
+ // then every check skips and the run exits 0 -- SKIPPED is silence, not a failure signal
+ assertEquals(HoodieTableHealthChecker.EXIT_HEALTHY, result.exitCode);
+ assertTrue(result.output.contains("SKIPPED"), result.output);
+ assertTrue(result.output.contains("--apply-all-defaults"), result.output);
+ }
+
+ @Test
+ public void applyAllDefaultsTurnsTheSameTableUnhealthy() throws Exception {
+ // given the same lagging table and still no supplied properties
+ laggingMorTable();
+ HoodieTableHealthChecker.Config cfg = configFor("all");
+ cfg.applyAllDefaults = true;
+
+ // when the operator explicitly accepts Hudi defaults
+ Result result = runChecker(cfg);
+
+ // then the checks actually evaluate, and the lag is caught
+ assertEquals(HoodieTableHealthChecker.EXIT_UNHEALTHY, result.exitCode);
+ }
+
+ @Test
+ public void checksCanBeSelectedIndividually() throws Exception {
+ // given any table
+ healthyMorTable();
+ HoodieTableHealthChecker.Config cfg = configFor("archival");
+ cfg.configs = Arrays.asList(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key() + "=30");
+
+ // when only one check is requested
+ Result result = runChecker(cfg);
+
+ // then only that check is reported
+ assertTrue(result.output.contains("archival"), result.output);
+ assertTrue(!result.output.contains("] compaction"), result.output);
+ }
+
+ @Test
+ public void unknownCheckNameIsRejectedWithTheAvailableNames() throws Exception {
+ // given a request for a check that does not exist
+ healthyMorTable();
+
+ // when the checker runs
+ IllegalArgumentException thrown = assertThrows(IllegalArgumentException.class,
+ () -> runChecker(configFor("no-such-check")));
+
+ // then the error names the valid options rather than failing opaquely
+ assertTrue(thrown.getMessage().contains("no-such-check"), thrown.getMessage());
+ assertTrue(thrown.getMessage().contains("compaction"), thrown.getMessage());
+ }
+
+ @Test
+ public void jsonOutputIsMachineReadableAndCarriesEveryVerdict() throws Exception {
+ // given a lagging table
+ laggingMorTable();
+ HoodieTableHealthChecker.Config cfg = configFor("all");
+ cfg.output = HoodieTableHealthChecker.OutputFormat.JSON;
+ cfg.configs = Arrays.asList(
+ HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.key() + "=NUM_COMMITS",
+ HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key() + "=5",
+ HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key() + "=1",
+ HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key() + "=10");
+
+ // when the checker runs with JSON output
+ Result result = runChecker(cfg);
+
+ // then the report parses, and every check appears with its status and effective config
+ 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());
+
+ JsonNode compaction = root.get("checks").get(0);
+ assertEquals("compaction", compaction.get("name").asText());
+ assertEquals("UNHEALTHY", compaction.get("status").asText());
+ assertTrue(compaction.get("findings").size() > 0);
+ assertTrue(compaction.get("effectiveConfigs")
+ .has(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key()));
+ }
+
+ @Test
+ public void verboseOutputEchoesTheConfigEachCheckUsed() throws Exception {
+ // given a healthy table checked verbosely
+ healthyMorTable();
+ HoodieTableHealthChecker.Config cfg = configFor("archival");
+ cfg.verbose = true;
+ cfg.configs = Arrays.asList(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key() + "=30");
+
+ // when the checker runs
+ Result result = runChecker(cfg);
+
+ // then the verdict can be audited against the values that produced it
+ assertTrue(result.output.contains("config used:"), result.output);
+ assertTrue(result.output.contains(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key()), result.output);
+ }
+
+ private static class Result {
+ final int exitCode;
+ final String output;
+
+ Result(int exitCode, String output) {
+ this.exitCode = exitCode;
+ this.output = output;
+ }
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestArchivalCadenceCheck.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestArchivalCadenceCheck.java
new file mode 100644
index 000000000000..d46f4e8b6e33
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestArchivalCadenceCheck.java
@@ -0,0 +1,185 @@
+/*
+ * 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.TypedProperties;
+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.common.testutils.HoodieTestUtils;
+import org.apache.hudi.config.HoodieArchivalConfig;
+import org.apache.hudi.config.HoodieCleanConfig;
+
+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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link ArchivalCadenceCheck}.
+ */
+public class TestArchivalCadenceCheck {
+
+ @TempDir
+ Path tempDir;
+
+ private String basePath;
+ private final ArchivalCadenceCheck check = new ArchivalCadenceCheck();
+
+ @BeforeEach
+ public void setUp() {
+ basePath = tempDir.resolve("table").toString();
+ }
+
+ private TypedProperties propsWithRetention(int maxCommitsToKeep) {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), String.valueOf(maxCommitsToKeep));
+ return props;
+ }
+
+ private HealthCheckResult runCheck(HoodieTableMetaClient metaClient, TypedProperties props) throws Exception {
+ return check.check(new HealthCheckContext(HoodieTableMetaClient.reload(metaClient), props, false));
+ }
+
+ private HoodieTableMetaClient tableWithCommits(int numCommits) throws Exception {
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = 1; i <= numCommits; i++) {
+ table.addCommit(String.format("%05d", i));
+ }
+ return metaClient;
+ }
+
+ @Test
+ public void missingRetentionConfigSkipsRatherThanGuessing() throws Exception {
+ // given a table with a long active timeline
+ HoodieTableMetaClient metaClient = tableWithCommits(50);
+
+ // when the check runs without the archival retention configured
+ HealthCheckResult result = runCheck(metaClient, new TypedProperties());
+
+ // then it declines to judge instead of calling a possibly fine table unhealthy
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key()));
+ }
+
+ @Test
+ public void timelineWithinRetentionIsHealthy() throws Exception {
+ // given a timeline comfortably inside the configured retention
+ HoodieTableMetaClient metaClient = tableWithCommits(5);
+
+ // when the check runs with a retention of 30
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(30));
+
+ // then archival is keeping up
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ }
+
+ @Test
+ public void timelineSlightlyOverRetentionIsStillHealthy() throws Exception {
+ // given 10 write instants against a retention of 10 -- archival runs periodically, not per commit
+ HoodieTableMetaClient metaClient = tableWithCommits(10);
+
+ // when the check runs (threshold is 11 after the 1.1 slack factor)
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(10));
+
+ // then the normal gap between archival runs is not an alert
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ }
+
+ @Test
+ public void timelineWellPastRetentionIsUnhealthy() throws Exception {
+ // given far more write instants than the configured retention allows
+ HoodieTableMetaClient metaClient = tableWithCommits(60);
+
+ // when the check runs with a retention of 10
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(10));
+
+ // then archival is reported as not keeping up
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("not keeping up with ingestion")),
+ "expected a retention finding, got: " + result.getFindings());
+ }
+
+ @Test
+ public void unhealthyResultPointsAtTheSavepointCheckFirst() throws Exception {
+ // given a table whose archival has stalled
+ HoodieTableMetaClient metaClient = tableWithCommits(60);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(10));
+
+ // then it directs the operator to the most common cause rather than guessing at one itself
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains(SavepointArchivalBlockCheck.NAME)),
+ "expected a pointer to the savepoint check, got: " + result.getFindings());
+ }
+
+ @Test
+ public void effectiveConfigIsEchoedSoAVerdictCanBeAudited() throws Exception {
+ // given a short timeline
+ HoodieTableMetaClient metaClient = tableWithCommits(3);
+
+ // when the check runs with a known retention
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(20));
+
+ // then both the configured value and the derived threshold are reported
+ assertEquals("20", result.getEffectiveConfigs().get(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key()));
+ assertEquals("22", result.getEffectiveConfigs().get("effective.unhealthy.threshold.completed.commits"));
+ assertEquals("3", result.getEffectiveConfigs().get("observed.completed.commits"));
+ }
+
+ @Test
+ public void cleanerRetentionRaisesTheBoundTheArchiverEnforces() throws Exception {
+ // given a table whose cleaner retains far more history than archival would on its own --
+ // ArchivalUtils raises the archiver's bounds to cover the cleaner, so ~100 instants is correct
+ HoodieTableMetaClient metaClient = tableWithCommits(100);
+ TypedProperties props = propsWithRetention(30);
+ props.setProperty(HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(), "20");
+ props.setProperty(HoodieCleanConfig.CLEANER_POLICY.key(), "KEEP_LATEST_COMMITS");
+ props.setProperty(HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key(), "80");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, props);
+
+ // then the table is not flagged merely for honouring its own cleaner configuration
+ assertEquals(HealthStatus.HEALTHY, result.getStatus(),
+ "a timeline held open by cleaner retention is not an archival failure");
+ assertTrue(Integer.parseInt(result.getEffectiveConfigs().get("effective.max.commits.to.keep")) > 30,
+ "the effective bound should be raised above the configured maximum");
+ }
+
+ @Test
+ public void pendingWriteInstantIsNamedAsTheBlocker() throws Exception {
+ // given a long timeline with an inflight commit that archival cannot move past
+ HoodieTableMetaClient metaClient = tableWithCommits(60);
+ HoodieTestTable.of(metaClient).addInflightCommit("00061");
+
+ // when the check runs against a small retention
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(10));
+
+ // then the finding names the instant to act on rather than only listing possible causes
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("00061")),
+ "expected the pending instant to be named, got: " + result.getFindings());
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestCleanerCadenceCheck.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestCleanerCadenceCheck.java
new file mode 100644
index 000000000000..696766639068
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestCleanerCadenceCheck.java
@@ -0,0 +1,329 @@
+/*
+ * 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.avro.model.HoodieActionInstant;
+import org.apache.hudi.avro.model.HoodieCleanerPlan;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.versioning.clean.CleanPlanV2MigrationHandler;
+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.config.HoodieArchivalConfig;
+import org.apache.hudi.config.HoodieCleanConfig;
+
+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.Collections;
+import java.util.HashMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link CleanerCadenceCheck}.
+ */
+public class TestCleanerCadenceCheck {
+
+ @TempDir
+ Path tempDir;
+
+ private String basePath;
+ private final CleanerCadenceCheck check = new CleanerCadenceCheck();
+
+ @BeforeEach
+ public void setUp() {
+ basePath = tempDir.resolve("table").toString();
+ }
+
+ /** A serializable but empty plan -- the Avro schema rejects the no-arg constructor's nulls. */
+ private static HoodieCleanerPlan emptyCleanerPlan() {
+ return new HoodieCleanerPlan(new HoodieActionInstant("", "", ""), "", "", new HashMap<>(),
+ CleanPlanV2MigrationHandler.VERSION, new HashMap<>(), new ArrayList<>(), Collections.emptyMap());
+ }
+
+ private TypedProperties propsWithTrigger(int triggerCommits) {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key(), String.valueOf(triggerCommits));
+ return props;
+ }
+
+ private HealthCheckResult runCheck(HoodieTableMetaClient metaClient, TypedProperties props,
+ boolean applyAllDefaults) throws Exception {
+ return check.check(new HealthCheckContext(HoodieTableMetaClient.reload(metaClient), props, applyAllDefaults));
+ }
+
+ private HoodieTestTable tableWithCommits(HoodieTableMetaClient metaClient, int from, int to) throws Exception {
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = from; i <= to; i++) {
+ table.addCommit(String.format("%05d", i));
+ }
+ return table;
+ }
+
+ @Test
+ public void missingTriggerConfigSkipsRatherThanGuessing() throws Exception {
+ // given a table with many commits and no clean
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ tableWithCommits(metaClient, 1, 30);
+
+ // when the check runs without the cleaner trigger configured
+ HealthCheckResult result = runCheck(metaClient, new TypedProperties(), false);
+
+ // then it declines to judge and names what it needs
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains(HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key()));
+ }
+
+ @Test
+ public void applyAllDefaultsEvaluatesWithoutSuppliedConfig() throws Exception {
+ // given the same never-cleaned table
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ tableWithCommits(metaClient, 1, 30);
+
+ // when the operator explicitly opts into Hudi defaults (trigger of 1)
+ HealthCheckResult result = runCheck(metaClient, new TypedProperties(), true);
+
+ // then the check runs and reports the cleaner as badly behind
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ }
+
+ @Test
+ public void fewCommitsWithoutACleanAreStillHealthy() throws Exception {
+ // given a young table with three commits and no clean yet
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ tableWithCommits(metaClient, 1, 3);
+
+ // when the check runs with a trigger of 1 (threshold floors at 5)
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(1), false);
+
+ // then ordinary async lag is not an alert
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ assertEquals("none", result.getEffectiveConfigs().get("observed.last.clean.instant"));
+ }
+
+ @Test
+ public void thresholdFloorsAtMinimumEvenForATriggerOfOne() throws Exception {
+ // given any table
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ tableWithCommits(metaClient, 1, 1);
+
+ // when the check runs with the lowest possible trigger
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(1), false);
+
+ // then the threshold is the floor, not 2
+ assertEquals(String.valueOf(CleanerCadenceCheck.MIN_UNHEALTHY_COMMITS),
+ result.getEffectiveConfigs().get("effective.unhealthy.threshold.commits"));
+ }
+
+ @Test
+ public void neverCleanedTableWithManyCommitsIsUnhealthy() throws Exception {
+ // given a table that has taken many commits and never been cleaned
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ tableWithCommits(metaClient, 1, 30);
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(1), false);
+
+ // then it is unhealthy and says nothing is scheduled
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getSummary().contains("never been cleaned"), result.getSummary());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("No clean is scheduled")),
+ "expected a not-scheduled finding, got: " + result.getFindings());
+ }
+
+ @Test
+ public void commitsAfterACompletedCleanResetTheCount() throws Exception {
+ // given a table that fell behind, cleaned, then took two more commits
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = tableWithCommits(metaClient, 1, 30);
+ table.addClean("00031");
+ table.addCommit("00032").addCommit("00033");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(1), false);
+
+ // then only the commits after that clean count
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ assertEquals("00031", result.getEffectiveConfigs().get("observed.last.clean.instant"));
+ assertEquals("2", result.getEffectiveConfigs().get("observed.commits.since.last.clean"));
+ }
+
+ @Test
+ public void staleCleanWithManyCommitsAfterItIsUnhealthy() throws Exception {
+ // given a table whose last clean is long behind the head of the timeline
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = tableWithCommits(metaClient, 1, 2);
+ table.addClean("00003");
+ for (int i = 4; i <= 40; i++) {
+ table.addCommit(String.format("%05d", i));
+ }
+
+ // when the check runs with a trigger of 3 (threshold 6)
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(3), false);
+
+ // then the cleaner is reported as behind, naming the last clean
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getSummary().contains("00003"), result.getSummary());
+ assertEquals("37", result.getEffectiveConfigs().get("observed.commits.since.last.clean"));
+ }
+
+ @Test
+ public void inflightCleanIsCalledOutAsSchedulingWithoutCompletion() throws Exception {
+ // given a lagging table with a clean requested but never finished
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = tableWithCommits(metaClient, 1, 30);
+ table.addInflightClean("00031", emptyCleanerPlan());
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(1), false);
+
+ // then the diagnosis distinguishes "not scheduled" from "scheduled but stuck"
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("has not completed")),
+ "expected a stuck-clean finding, got: " + result.getFindings());
+ }
+
+ @Test
+ public void optionalPolicyConfigIsEchoedOnlyWhenSupplied() throws Exception {
+ // given a table and a props set that includes the retention policy
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ tableWithCommits(metaClient, 1, 1);
+ TypedProperties props = propsWithTrigger(1);
+ props.setProperty(HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key(), "20");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, props, false);
+
+ // then the supplied value is echoed and the unsupplied policy is not invented
+ assertEquals("20", result.getEffectiveConfigs().get(HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key()));
+ assertTrue(!result.getEffectiveConfigs().containsKey(HoodieCleanConfig.CLEANER_POLICY.key()));
+ }
+
+ @Test
+ public void archivalStarvingTheCleanerIsNamedInsteadOfBlamingAutoClean() throws Exception {
+ // given keep.min.commits at or below clean.commits.retained: archival trims the timeline
+ // before the cleaner ever sees enough commits, so no clean is scheduled by construction
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ tableWithCommits(metaClient, 1, 60);
+ TypedProperties props = propsWithTrigger(1);
+ props.setProperty(HoodieCleanConfig.CLEANER_POLICY.key(), "KEEP_LATEST_COMMITS");
+ props.setProperty(HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key(), "30");
+ props.setProperty(HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(), "20");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, props, false);
+
+ // then the finding points at the config pairing, not at hoodie.clean.automatic
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getFindings().stream()
+ .anyMatch(f -> f.contains(HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key())
+ && f.contains(HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key())),
+ "expected the archival/cleaner conflict to be named, got: " + result.getFindings());
+ assertTrue(result.getFindings().stream().noneMatch(f -> f.startsWith("No clean is scheduled.")),
+ "should not fall back to the generic finding when the configs explain it");
+ }
+
+ @Test
+ public void healthyArchivalPairingStillGivesTheGenericFinding() throws Exception {
+ // given keep.min.commits comfortably above clean.commits.retained, the supported pairing
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ tableWithCommits(metaClient, 1, 60);
+ TypedProperties props = propsWithTrigger(1);
+ props.setProperty(HoodieCleanConfig.CLEANER_POLICY.key(), "KEEP_LATEST_COMMITS");
+ props.setProperty(HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key(), "10");
+ props.setProperty(HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(), "20");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, props, false);
+
+ // then the configuration is not blamed, and the operator is pointed at the service
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.startsWith("No clean is scheduled.")),
+ "expected the generic finding, got: " + result.getFindings());
+ }
+
+ @Test
+ public void appendOnlyTableWithNoCleanIsHealthy() throws Exception {
+ // given a table that has only ever received inserts -- nothing is superseded, so the
+ // cleaner correctly never runs and no clean instant ever appears
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = 1; i <= 60; i++) {
+ String instant = String.format("%05d", i);
+ table.addCommit(instant, Option.of(table.createCommitMetadata(
+ instant, WriteOperationType.INSERT, Collections.singletonList("p1"), 1, false)));
+ }
+
+ // when the check runs well past the trigger threshold
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(1), false);
+
+ // then an empty cleaner timeline is the expected state, not a stalled service
+ assertEquals(HealthStatus.HEALTHY, result.getStatus(),
+ "an append-only table has nothing to clean: " + result.getFindings());
+ assertTrue(result.getSummary().contains("none is expected"),
+ "expected the summary to say why, got: " + result.getSummary());
+ }
+
+ @Test
+ public void upsertTableWithNoCleanIsStillJudged() throws Exception {
+ // given the same timeline length, but writes that supersede file slices
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = 1; i <= 60; i++) {
+ String instant = String.format("%05d", i);
+ table.addCommit(instant, Option.of(table.createCommitMetadata(
+ instant, WriteOperationType.UPSERT, Collections.singletonList("p1"), 1, false)));
+ }
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(1), false);
+
+ // then the append-only exemption must not swallow a genuine finding
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus(),
+ "an upsert table with no clean is a real finding");
+ }
+
+ @Test
+ public void mixedInsertAndUpsertIsStillJudged() throws Exception {
+ // given a table that is mostly inserts but took updates at some point
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = 1; i <= 60; i++) {
+ String instant = String.format("%05d", i);
+ WriteOperationType op = (i == 30) ? WriteOperationType.UPSERT : WriteOperationType.INSERT;
+ table.addCommit(instant, Option.of(table.createCommitMetadata(
+ instant, op, Collections.singletonList("p1"), 1, false)));
+ }
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(1), false);
+
+ // then one update is enough to make cleaning meaningful again
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus(),
+ "a single upsert means file slices can be superseded");
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestCompactionCadenceCheck.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestCompactionCadenceCheck.java
new file mode 100644
index 000000000000..83b103b9d41a
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestCompactionCadenceCheck.java
@@ -0,0 +1,213 @@
+/*
+ * 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.TypedProperties;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+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.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieCompactionConfig;
+
+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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link CompactionCadenceCheck} against real timelines built with {@link HoodieTestTable}.
+ */
+public class TestCompactionCadenceCheck {
+
+ @TempDir
+ Path tempDir;
+
+ private String basePath;
+ private final CompactionCadenceCheck check = new CompactionCadenceCheck();
+
+ @BeforeEach
+ public void setUp() {
+ basePath = tempDir.resolve("table").toString();
+ }
+
+ private HoodieTableMetaClient initTable(HoodieTableType tableType) throws Exception {
+ return HoodieTestUtils.init(basePath, tableType);
+ }
+
+ private TypedProperties propsWithTrigger(int maxDeltaCommits) {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.key(), "NUM_COMMITS");
+ props.setProperty(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(),
+ String.valueOf(maxDeltaCommits));
+ return props;
+ }
+
+ private HealthCheckResult runCheck(HoodieTableMetaClient metaClient, TypedProperties props,
+ boolean applyAllDefaults) throws Exception {
+ // Reload so the check sees every instant written after the meta client was created.
+ HoodieTableMetaClient reloaded = HoodieTableMetaClient.reload(metaClient);
+ return check.check(new HealthCheckContext(reloaded, props, applyAllDefaults));
+ }
+
+ @Test
+ public void copyOnWriteTableIsSkippedBecauseItHasNoCompaction() throws Exception {
+ // given a Copy-on-Write table
+ HoodieTableMetaClient metaClient = initTable(HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable.of(metaClient).addCommit("001").addCommit("002");
+
+ // when the compaction check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(5), false);
+
+ // then it is skipped rather than reported either way
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains("Copy-on-Write"));
+ }
+
+ @Test
+ public void missingTriggerConfigSkipsRatherThanAssumingDefaults() throws Exception {
+ // given a MOR table with delta commits piled up well past the stock trigger
+ HoodieTableMetaClient metaClient = initTable(HoodieTableType.MERGE_ON_READ);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = 1; i <= 40; i++) {
+ table.addDeltaCommit(String.format("%03d", i));
+ }
+
+ // when the check runs without the writer's trigger configuration
+ HealthCheckResult result = runCheck(metaClient, new TypedProperties(), false);
+
+ // then it declines to judge, naming what it needs, instead of calling the table unhealthy
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains(HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.key()));
+ }
+
+ @Test
+ public void applyAllDefaultsEvaluatesWithoutSuppliedConfig() throws Exception {
+ // given the same table and still no supplied configuration
+ HoodieTableMetaClient metaClient = initTable(HoodieTableType.MERGE_ON_READ);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = 1; i <= 40; i++) {
+ table.addDeltaCommit(String.format("%03d", i));
+ }
+
+ // when the operator explicitly opts into Hudi defaults
+ HealthCheckResult result = runCheck(metaClient, new TypedProperties(), true);
+
+ // then the check runs and, with 40 delta commits against a default trigger of 5, reports badly behind
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ }
+
+ @Test
+ public void deltaCommitsWithinSlackAreHealthy() throws Exception {
+ // given a MOR table whose delta commits sit inside the trigger threshold plus slack
+ HoodieTableMetaClient metaClient = initTable(HoodieTableType.MERGE_ON_READ);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ table.addDeltaCommit("001").addDeltaCommit("002").addDeltaCommit("003");
+
+ // when the check runs with a trigger of 5 (threshold 10 after the 2.0 slack factor)
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(5), false);
+
+ // then the table is healthy -- scheduling jitter is not an alert
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ }
+
+ @Test
+ public void deltaCommitsPastSlackAreUnhealthyAndSayNoCompactionIsScheduled() throws Exception {
+ // given a MOR table with delta commits far past the trigger and nothing scheduled
+ HoodieTableMetaClient metaClient = initTable(HoodieTableType.MERGE_ON_READ);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = 1; i <= 25; i++) {
+ table.addDeltaCommit(String.format("%03d", i));
+ }
+
+ // when the check runs with a trigger of 5 (threshold 10)
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(5), false);
+
+ // then it reports unhealthy and points at compaction not being scheduled at all
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("No compaction is scheduled")),
+ "expected a finding about compaction not being scheduled, got: " + result.getFindings());
+ }
+
+ @Test
+ public void pendingCompactionIsCalledOutAsSchedulingWithoutExecution() throws Exception {
+ // given a MOR table behind on compaction, with a plan requested but never completed
+ HoodieTableMetaClient metaClient = initTable(HoodieTableType.MERGE_ON_READ);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = 1; i <= 25; i++) {
+ table.addDeltaCommit(String.format("%03d", i));
+ }
+ table.addRequestedCompaction("026");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(5), false);
+
+ // then the diagnosis distinguishes "not scheduled" from "scheduled but not executed"
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("not executed")),
+ "expected a finding about compaction not executing, got: " + result.getFindings());
+ }
+
+ @Test
+ public void deltaCommitsAfterACompletedCompactionResetTheCount() throws Exception {
+ // given a MOR table that fell behind, then compacted, then took a few more delta commits.
+ // Completion times are set explicitly and kept in step with requested times: the underlying
+ // timeline lookup selects delta commits by *completion* time, so a fixture that leaves them
+ // unset does not model a real timeline.
+ HoodieTableMetaClient metaClient = initTable(HoodieTableType.MERGE_ON_READ);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ for (int i = 1; i <= 25; i++) {
+ String instant = String.format("%03d", i);
+ table.addDeltaCommit(instant, Option.of(instant), new HoodieCommitMetadata());
+ }
+ table.addCompaction("026", Option.of("026"), new HoodieCommitMetadata());
+ table.addDeltaCommit("027", Option.of("027"), new HoodieCommitMetadata());
+ table.addDeltaCommit("028", Option.of("028"), new HoodieCommitMetadata());
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(5), false);
+
+ // then only the two delta commits after that compaction count, so the table reads healthy
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ assertEquals("026", result.getEffectiveConfigs().get("observed.last.compaction.instant"));
+ assertEquals("2", result.getEffectiveConfigs().get("observed.delta.commits.since.last.compaction"));
+ }
+
+ @Test
+ public void effectiveConfigIsEchoedSoAVerdictCanBeAudited() throws Exception {
+ // given any MOR table
+ HoodieTableMetaClient metaClient = initTable(HoodieTableType.MERGE_ON_READ);
+ HoodieTestTable.of(metaClient).addDeltaCommit("001");
+
+ // when the check runs with a known trigger
+ HealthCheckResult result = runCheck(metaClient, propsWithTrigger(7), false);
+
+ // then the values it reasoned with are reported alongside the verdict
+ assertEquals("7", result.getEffectiveConfigs()
+ .get(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key()));
+ assertEquals("NUM_COMMITS", result.getEffectiveConfigs()
+ .get(HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY.key()));
+ assertEquals("14", result.getEffectiveConfigs().get("effective.unhealthy.threshold.delta.commits"));
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestHealthCheckContext.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestHealthCheckContext.java
new file mode 100644
index 000000000000..24c06f07c7c7
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestHealthCheckContext.java
@@ -0,0 +1,103 @@
+/*
+ * 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.TypedProperties;
+
+import org.junit.jupiter.api.Test;
+
+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 the fail-fast configuration contract: a check may only reason with writer properties the
+ * operator actually supplied, unless defaults were explicitly opted into.
+ */
+public class TestHealthCheckContext {
+
+ private static final String KEY_A = "hoodie.some.property";
+ private static final String KEY_B = "hoodie.other.property";
+
+ private HealthCheckContext contextWith(boolean applyAllDefaults, String... keys) {
+ TypedProperties props = new TypedProperties();
+ for (String key : keys) {
+ props.setProperty(key, "value");
+ }
+ return new HealthCheckContext(null, props, applyAllDefaults);
+ }
+
+ @Test
+ public void suppliedConfigsAreUsable() {
+ // given a context where both properties were supplied
+ HealthCheckContext context = contextWith(false, KEY_A, KEY_B);
+
+ // when asked whether it may reason with them
+ // then it may, and nothing is reported missing
+ assertTrue(context.hasConfigsOrDefaultsAllowed(KEY_A, KEY_B));
+ assertFalse(context.firstMissingConfig(KEY_A, KEY_B).isPresent());
+ }
+
+ @Test
+ public void missingConfigIsReportedRatherThanDefaulted() {
+ // given a context where only one of two required properties was supplied
+ HealthCheckContext context = contextWith(false, KEY_A);
+
+ // when asked about both
+ // then the unsupplied one is named, rather than silently taking a default
+ assertFalse(context.hasConfigsOrDefaultsAllowed(KEY_A, KEY_B));
+ assertEquals(KEY_B, context.firstMissingConfig(KEY_A, KEY_B).get());
+ }
+
+ @Test
+ public void firstMissingConfigNamesTheEarliestGap() {
+ // given a context with none of the required properties
+ HealthCheckContext context = contextWith(false);
+
+ // when asked about several
+ // then the first is named, so the operator has one concrete thing to fix
+ assertEquals(KEY_A, context.firstMissingConfig(KEY_A, KEY_B).get());
+ }
+
+ @Test
+ public void applyAllDefaultsPermitsMissingConfigs() {
+ // given a context that explicitly opted into Hudi defaults
+ HealthCheckContext context = contextWith(true);
+
+ // when asked about properties nobody supplied
+ // then they are treated as available, because the operator accepted the risk
+ assertTrue(context.hasConfigsOrDefaultsAllowed(KEY_A, KEY_B));
+ assertFalse(context.firstMissingConfig(KEY_A, KEY_B).isPresent());
+ }
+
+ @Test
+ public void skipResultNamesThePropertyAndBothWaysForward() {
+ // given a context missing a property
+ HealthCheckContext context = contextWith(false);
+
+ // when a check skips because of it
+ HealthCheckResult result = context.skipForMissingConfig("some-check", KEY_A);
+
+ // then the operator is told what is missing and both ways to proceed
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains(KEY_A));
+ assertTrue(result.getSummary().contains("--props"));
+ assertTrue(result.getSummary().contains("--apply-all-defaults"));
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestHealthStatus.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestHealthStatus.java
new file mode 100644
index 000000000000..b020e1f6670c
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestHealthStatus.java
@@ -0,0 +1,62 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+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 the severity ordering that table-level verdicts are rolled up with.
+ */
+public class TestHealthStatus {
+
+ @Test
+ public void unhealthyDominatesEveryOtherStatus() {
+ // given any pairing with UNHEALTHY
+ // then UNHEALTHY wins, so one bad check is never masked by good ones
+ assertEquals(HealthStatus.UNHEALTHY, HealthStatus.max(HealthStatus.UNHEALTHY, HealthStatus.HEALTHY));
+ assertEquals(HealthStatus.UNHEALTHY, HealthStatus.max(HealthStatus.HEALTHY, HealthStatus.UNHEALTHY));
+ assertEquals(HealthStatus.UNHEALTHY, HealthStatus.max(HealthStatus.UNHEALTHY, HealthStatus.SKIPPED));
+ }
+
+ @Test
+ public void healthyOutranksSkipped() {
+ // given a run where one check passed and another was skipped
+ // then the table reads HEALTHY: a skipped check is silent, not a failure
+ assertEquals(HealthStatus.HEALTHY, HealthStatus.max(HealthStatus.SKIPPED, HealthStatus.HEALTHY));
+ assertEquals(HealthStatus.HEALTHY, HealthStatus.max(HealthStatus.HEALTHY, HealthStatus.SKIPPED));
+ }
+
+ @Test
+ public void allSkippedRollsUpToSkipped() {
+ // given a run where nothing could be evaluated
+ // then the verdict is SKIPPED rather than a false clean bill of health
+ assertEquals(HealthStatus.SKIPPED, HealthStatus.max(HealthStatus.SKIPPED, HealthStatus.SKIPPED));
+ }
+
+ @Test
+ public void onlyUnhealthyReadsAsUnhealthy() {
+ assertTrue(HealthStatus.UNHEALTHY.isUnhealthy());
+ assertFalse(HealthStatus.HEALTHY.isUnhealthy());
+ assertFalse(HealthStatus.SKIPPED.isUnhealthy());
+ }
+}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestSavepointArchivalBlockCheck.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestSavepointArchivalBlockCheck.java
new file mode 100644
index 000000000000..ab4357cdc70f
--- /dev/null
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/health/TestSavepointArchivalBlockCheck.java
@@ -0,0 +1,190 @@
+/*
+ * 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.TypedProperties;
+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.common.testutils.HoodieTestUtils;
+import org.apache.hudi.config.HoodieArchivalConfig;
+
+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.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests {@link SavepointArchivalBlockCheck}.
+ *
+ * <p>Archival stops at the earliest savepoint, so the behaviour under test is unconditional: a
+ * savepoint sitting behind the archival retention pins the timeline, and the only remedy offered is
+ * releasing it.
+ */
+public class TestSavepointArchivalBlockCheck {
+
+ private static final DateTimeFormatter INSTANT_FORMAT =
+ DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS").withZone(ZoneId.systemDefault());
+
+ @TempDir
+ Path tempDir;
+
+ private String basePath;
+ private final SavepointArchivalBlockCheck check = new SavepointArchivalBlockCheck();
+
+ @BeforeEach
+ public void setUp() {
+ basePath = tempDir.resolve("table").toString();
+ }
+
+ /** An instant time the given number of days in the past, parseable by the staleness check. */
+ private static String instantDaysAgo(int daysAgo) {
+ return INSTANT_FORMAT.format(Instant.now().minus(daysAgo, ChronoUnit.DAYS));
+ }
+
+ private TypedProperties propsWithRetention(int maxCommitsToKeep) {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), String.valueOf(maxCommitsToKeep));
+ return props;
+ }
+
+ private HealthCheckResult runCheck(HoodieTableMetaClient metaClient, TypedProperties props) throws Exception {
+ return check.check(new HealthCheckContext(HoodieTableMetaClient.reload(metaClient), props, false));
+ }
+
+ @Test
+ public void tableWithNoSavepointsIsHealthy() throws Exception {
+ // given a table that has never been savepointed
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable.of(metaClient).addCommit("001").addCommit("002");
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(10));
+
+ // then nothing is blocking archival
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ assertEquals("0", result.getEffectiveConfigs().get("observed.savepoint.count"));
+ }
+
+ @Test
+ public void missingRetentionConfigSkipsRatherThanGuessing() throws Exception {
+ // given a table with a savepoint
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ String savepointed = instantDaysAgo(30);
+ table.addCommit(savepointed);
+ table.addSavepoint(savepointed, table.getSavepointMetadata(savepointed, Collections.emptyMap()));
+
+ // when the check runs without the archival retention configured
+ HealthCheckResult result = runCheck(metaClient, new TypedProperties());
+
+ // then it declines to judge and names what it needs
+ assertEquals(HealthStatus.SKIPPED, result.getStatus());
+ assertTrue(result.getSummary().contains(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key()));
+ }
+
+ @Test
+ public void recentSavepointInsideRetentionIsHealthy() throws Exception {
+ // given a small table whose only savepoint is well inside the retention window
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ String savepointed = instantDaysAgo(1);
+ table.addCommit(savepointed);
+ table.addSavepoint(savepointed, table.getSavepointMetadata(savepointed, Collections.emptyMap()));
+
+ // when the check runs with a retention larger than the timeline
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(10));
+
+ // then archival is not yet blocked by it
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ }
+
+ @Test
+ public void savepointBehindRetentionBlocksArchival() throws Exception {
+ // given a savepoint on the oldest commit, with many commits written after it
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ String savepointed = instantDaysAgo(60);
+ table.addCommit(savepointed);
+ table.addSavepoint(savepointed, table.getSavepointMetadata(savepointed, Collections.emptyMap()));
+ for (int i = 1; i <= 20; i++) {
+ table.addCommit(instantDaysAgo(60 - i));
+ }
+
+ // when the check runs with a retention of 5 commits
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(5));
+
+ // then the savepoint is reported as blocking archival, naming the instant
+ assertEquals(HealthStatus.UNHEALTHY, result.getStatus());
+ assertTrue(result.getSummary().contains(savepointed));
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("Archival stops at the earliest savepoint")),
+ "expected the archival-stops finding, got: " + result.getFindings());
+ }
+
+ @Test
+ public void remedyIsAlwaysToReleaseTheSavepoint() throws Exception {
+ // given a savepoint blocking archival
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ String savepointed = instantDaysAgo(60);
+ table.addCommit(savepointed);
+ table.addSavepoint(savepointed, table.getSavepointMetadata(savepointed, Collections.emptyMap()));
+ for (int i = 1; i <= 20; i++) {
+ table.addCommit(instantDaysAgo(60 - i));
+ }
+
+ // when the check runs
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(5));
+
+ // then the advice is to delete the savepoint, and never to loosen archival past savepoints
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("savepoint delete")),
+ "expected advice to delete the savepoint, got: " + result.getFindings());
+ assertTrue(result.getFindings().stream().noneMatch(f -> f.contains("beyond.savepoint")),
+ "must not suggest archiving beyond savepoints: " + result.getFindings());
+ assertTrue(result.getEffectiveConfigs().keySet().stream().noneMatch(k -> k.contains("beyond.savepoint")),
+ "must not report a config this tool does not rely on: " + result.getEffectiveConfigs());
+ }
+
+ @Test
+ public void staleSavepointIsFlaggedEvenWhenItIsNotYetBlocking() throws Exception {
+ // given a long-forgotten savepoint on a table too short for it to block archival yet
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(basePath, HoodieTableType.COPY_ON_WRITE);
+ HoodieTestTable table = HoodieTestTable.of(metaClient);
+ String savepointed = instantDaysAgo(90);
+ table.addCommit(savepointed);
+ table.addSavepoint(savepointed, table.getSavepointMetadata(savepointed, Collections.emptyMap()));
+
+ // when the check runs with a retention larger than the timeline
+ HealthCheckResult result = runCheck(metaClient, propsWithRetention(50));
+
+ // then archival is not blocked, but the forgotten savepoint is still surfaced
+ assertEquals(HealthStatus.HEALTHY, result.getStatus());
+ assertTrue(result.getFindings().stream().anyMatch(f -> f.contains("days old")),
+ "expected a staleness finding, got: " + result.getFindings());
+ }
+}