hudi-agent commented on code in PR #20151: URL: https://github.com/apache/hudi/pull/20151#discussion_r4140597632
########## 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; Review Comment: 🤖 For TIME_ELAPSED (and NUM_OR_TIME through the `min` on L196), if no delta commit landed in the last `max.delta.seconds` (1h by default), the expected count is 0. The threshold is then 0 and a single pending delta commit reads UNHEALTHY, so a daily batch MOR table checked a few hours after its run would be flagged every day. Could this mirror the scheduler instead? For example, flag when `latestDeltaCommit.requestedTime - lastCompaction.requestedTime` is more than the slack factor times `max.delta.seconds`. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/CleanerCadenceCheck.java: ########## @@ -0,0 +1,151 @@ +/* + * 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.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.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()); + + int commitsSinceClean; + if (lastClean.isPresent()) { Review Comment: 🤖 By default Hudi doesn't write a clean instant when the plan is empty (`hoodie.write.empty.clean.interval.hours` defaults to -1). An append-only or mostly-insert table (bulk_insert, or updates rarer than every 5 commits) can go dozens of commits without a completed clean while the cleaner runs fine, and this check would call it UNHEALTHY with "No clean is scheduled". Is there a way to tell "cleaner not running" apart from "nothing to clean"? Even just mentioning empty cleans in the finding would help. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## 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; Review Comment: 🤖 nit: this class otherwise uses `java.time.Instant`/`ChronoUnit` for date math — using `java.sql.Date` here (via `TimelineUtils.formatDate`) mixes two date APIs in the same file and could confuse a future reader about which one to use for new date logic. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## 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; Review Comment: 🤖 With NUM_COMMITS_AFTER_LAST_REQUEST, the budget grows with the backlog. Suppose the compactor stops but the scheduler keeps adding plans (the normal pattern for this strategy, e.g. Flink). Then `between` covers nearly every delta commit since the last completed compaction, and `2 * (between + max)` always stays above the observed count. The check can never report the "scheduled but not executed" case it's meant to catch. Would it work to judge against `max` since the last completed compaction, or to flag on the number of pending plans? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java: ########## @@ -0,0 +1,327 @@ +/* + * 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.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 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 1} -- at least one check reported unhealthy</li> + * <li>{@code 2} -- bad arguments, or the table could not be read</li> + * </ul> + * + * <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; + static final int EXIT_UNHEALTHY = 1; + static final int EXIT_ERROR = 2; + + private static final ObjectMapper JSON_MAPPER = + new ObjectMapper().enable(SerializationFeature.INDENT_OUTPUT); + + private final Config cfg; + private final TypedProperties props; + private final HoodieTableMetaClient metaClient; + + public HoodieTableHealthChecker(Configuration hadoopConf, Config cfg) { + this.cfg = cfg; + 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); + + 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)); + } + + if (!erroredChecks.isEmpty()) { + return EXIT_ERROR; + } + return results.stream().anyMatch(r -> r.getStatus().isUnhealthy()) ? EXIT_UNHEALTHY : 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 = {"--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" + + " --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); Review Comment: 🤖 `new JCommander(cfg, null, args)` parses and validates right here, outside the try. A missing `--base-path`, an unknown flag, or `--output XML` throws `ParameterException` out of `main`, so the JVM exits 1, which this tool uses to mean UNHEALTHY (the cron example would alert "table is unhealthy"). Could you wrap the parse and map `ParameterException` to `EXIT_ERROR`? Also, the `args.length == 0` branch can't be reached, because the required-parameter check throws first. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## 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(); Review Comment: 🤖 When the table has never compacted, `getCompletedDeltaCommitsSinceLatestCompaction` returns the *first delta commit* as the right side. The summary then reads "N delta commit(s) since compaction at 00001", and `observed.last.compaction.instant` names a delta commit (the JSON sample in the .md shows exactly this). Since the skill quotes summaries verbatim, could you check `lastCompaction.getAction()` and say "never compacted" in that case? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## 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( Review Comment: 🤖 The gating here doesn't quite match the "skip rather than guess" contract. For TIME_ELAPSED, NUM_OR_TIME and NUM_AND_TIME the verdict depends on `hoodie.compact.inline.max.delta.seconds`, but that key isn't gated, so it quietly falls back to 3600. Meanwhile `trigger.strategy` is required even though most writers never set it. Could the required keys depend on the strategy? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheckContext.java: ########## @@ -0,0 +1,110 @@ +/* + * 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; + +/** + * 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; + + public HealthCheckContext(HoodieTableMetaClient metaClient, TypedProperties props, boolean applyAllDefaults) { + this.metaClient = metaClient; + this.props = props; + this.applyAllDefaults = applyAllDefaults; + } + + public HoodieTableMetaClient getMetaClient() { + return metaClient; + } + + 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)) { Review Comment: 🤖 `props.containsKey(key)` (and the later `getString(key, default)` in the checks) only look at the primary key, so values supplied under a `withAlternatives` alias are missed. For example, with `hoodie.clean.max.commits` for `CLEAN_TRIGGER_MAX_COMMITS` the check SKIPs, and with `--apply-all-defaults` it evaluates against the default of 1 even though the operator supplied a value. Could this take the `ConfigProperty` and use `ConfigUtils.containsConfigProperty` / `getStringWithAltKeys`? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/HealthCheckContext.java: ########## @@ -0,0 +1,110 @@ +/* + * 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; + +/** + * 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; + + public HealthCheckContext(HoodieTableMetaClient metaClient, TypedProperties props, boolean applyAllDefaults) { + this.metaClient = metaClient; + this.props = props; + this.applyAllDefaults = applyAllDefaults; + } + + public HoodieTableMetaClient getMetaClient() { + return metaClient; + } + + 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) { Review Comment: 🤖 nit: `hasConfigsOrDefaultsAllowed` doesn't look like it's called from any of the four `HealthCheck` implementations (they all use `firstMissingConfig` instead) — only the test exercises it. Consider dropping it if it's not needed, or wire it in if it was meant to replace duplicated logic. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java: ########## @@ -0,0 +1,327 @@ +/* + * 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.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 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 1} -- at least one check reported unhealthy</li> + * <li>{@code 2} -- bad arguments, or the table could not be read</li> + * </ul> + * + * <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; + static final int EXIT_UNHEALTHY = 1; + static final int EXIT_ERROR = 2; + + private static final ObjectMapper JSON_MAPPER = + new ObjectMapper().enable(SerializationFeature.INDENT_OUTPUT); + + private final Config cfg; + private final TypedProperties props; + private final HoodieTableMetaClient metaClient; + + public HoodieTableHealthChecker(Configuration hadoopConf, Config cfg) { + this.cfg = cfg; + 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); + + List<HealthCheckResult> results = new ArrayList<>(); + List<String> erroredChecks = new ArrayList<>(); + + for (HealthCheck check : selected) { + try { + results.add(check.check(context)); Review Comment: 🤖 The `HealthCheck#appliesTo` javadoc says a check that doesn't apply is reported SKIPPED rather than run, but the runner never calls it. Only `CompactionCadenceCheck` behaves correctly, because it calls `appliesTo` itself. Since follow-up PRs will add more checks on this framework, could the runner honor it here? For example: `if (!check.appliesTo(context)) { results.add(HealthCheckResult.skipped(...)); continue; }`. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
