shangxinli commented on code in PR #19868:
URL: https://github.com/apache/hudi/pull/19868#discussion_r3991769875
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataMetrics.java:
##########
@@ -84,6 +85,22 @@ public class HoodieMetadataMetrics implements Serializable {
public static final String LOG_COMPACTION_FAILURES =
"logcompaction_failures";
public static final String PENDING_COMPACTIONS_FAILURES =
"pending_compactions_failures";
+ // Metadata table compaction health. The existing per-partition
baseFileCount/logFileCount gauges
+ // show the current shape of the metadata table, but not whether compaction
is keeping up with it.
+ // Completed delta commits on the metadata table since the last completed
compaction, sampled before
+ // table services run. On a healthy table this sawtooths up to
hoodie.metadata.compact.max.delta.commits
+ // and falls back after each compaction, so alert on a multiple of that
config rather than on the
+ // config value itself. A value that climbs past the peak and keeps going
means metadata table
+ // compaction is not being scheduled or is failing.
Review Comment:
Confirmed at both guards. I would rather not wire the Flink compaction
pipeline into this PR, so I have scoped the comment to table services driven
through `performTableServices` and called the streaming-write gap out
explicitly for a follow-up.
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataMetrics.java:
##########
@@ -164,6 +181,48 @@ public void updateSizeMetrics(HoodieTableMetaClient
metaClient, HoodieBackedTabl
for (Map.Entry<String, String> e : stats.entrySet()) {
setMetric(e.getKey(), Long.parseLong(e.getValue()));
}
+ long totalBaseFiles = sumStat(stats, metadataPartitions,
STAT_COUNT_BASE_FILES);
+ long totalLogFiles = sumStat(stats, metadataPartitions,
STAT_COUNT_LOG_FILES);
+ setMetric(STAT_TOTAL_BASE_FILE_COUNT, totalBaseFiles);
+ setMetric(STAT_TOTAL_LOG_FILE_COUNT, totalLogFiles);
+ setMetric(STAT_LOG_TO_BASE_FILE_RATIO_PERCENT,
logToBaseFileRatioPercent(totalBaseFiles, totalLogFiles));
+ }
+
+ /**
+ * Sums a per-partition stat, keyed as {@code <partition>.<statName>}, over
the enabled metadata partitions.
+ */
+ @VisibleForTesting
+ static long sumStat(Map<String, String> stats, Set<String>
metadataPartitions, String statName) {
+ return metadataPartitions.stream()
+ .mapToLong(partition -> Long.parseLong(stats.getOrDefault(partition +
"." + statName, "0")))
Review Comment:
Right, and moot now: `sumStat` and both of those tests went away with the
size gauges.
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataMetrics.java:
##########
@@ -84,6 +85,22 @@ public class HoodieMetadataMetrics implements Serializable {
public static final String LOG_COMPACTION_FAILURES =
"logcompaction_failures";
public static final String PENDING_COMPACTIONS_FAILURES =
"pending_compactions_failures";
+ // Metadata table compaction health. The existing per-partition
baseFileCount/logFileCount gauges
Review Comment:
Fair observation, but it has been dead since 2020 and either wiring or
deleting it is a separate change from this one. An age-since-compaction gauge
is a good idea, and I would rather propose it on its own than fold it in here.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriter.java:
##########
@@ -651,4 +651,83 @@ void testPerformTableServicesWithFailureHandling(
// Verify metrics are incremented when there's a failure
verify(metrics,
times(1)).incrementMetric(HoodieMetadataMetrics.PENDING_COMPACTIONS_FAILURES,
1);
}
+
+ @Test
+ void performTableServicesReportsCompletedDeltaCommitsSinceLastCompaction()
throws Exception {
+ // A completed compaction, then three completed delta commits after it,
plus one still inflight.
+ // The gauge must count the three completed instants only, so that it
stays comparable to
+ // hoodie.compact.inline.max.delta.commits, which the compaction trigger
evaluates the same way.
+ List<HoodieInstant> instants = new ArrayList<>();
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
"001", "0011"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "002",
"0021"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
"003", "0031"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
"004", "0041"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
"005", "0051"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION,
"006"));
+
+ HoodieMetadataMetrics metrics = mock(HoodieMetadataMetrics.class);
+ writerForTableServices(metrics,
createMockTimeline(instants)).performTableServices(Option.empty(), true);
+
+ verify(metrics, times(1)).updateDeltaCommitsSinceLastCompaction(3L);
Review Comment:
Agreed, the old cases passed under `findInstantsAfter` too. Added a
`0015`/`0035` delta commit, requested before the compaction and completed after
it, plus a compaction-last case expecting 0.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriter.java:
##########
@@ -651,4 +651,83 @@ void testPerformTableServicesWithFailureHandling(
// Verify metrics are incremented when there's a failure
verify(metrics,
times(1)).incrementMetric(HoodieMetadataMetrics.PENDING_COMPACTIONS_FAILURES,
1);
}
+
+ @Test
+ void performTableServicesReportsCompletedDeltaCommitsSinceLastCompaction()
throws Exception {
+ // A completed compaction, then three completed delta commits after it,
plus one still inflight.
+ // The gauge must count the three completed instants only, so that it
stays comparable to
+ // hoodie.compact.inline.max.delta.commits, which the compaction trigger
evaluates the same way.
+ List<HoodieInstant> instants = new ArrayList<>();
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
"001", "0011"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "002",
"0021"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
"003", "0031"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
"004", "0041"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
"005", "0051"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION,
"006"));
+
+ HoodieMetadataMetrics metrics = mock(HoodieMetadataMetrics.class);
+ writerForTableServices(metrics,
createMockTimeline(instants)).performTableServices(Option.empty(), true);
+
+ verify(metrics, times(1)).updateDeltaCommitsSinceLastCompaction(3L);
+ }
+
+ @Test
+ void performTableServicesReportsAllDeltaCommitsWhenNeverCompacted() throws
Exception {
+ // No compaction has ever run, so every completed delta commit counts as
backlog.
+ List<HoodieInstant> instants = new ArrayList<>();
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
"001", "0011"));
+ instants.add(INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION,
"002", "0021"));
+
+ HoodieMetadataMetrics metrics = mock(HoodieMetadataMetrics.class);
+ writerForTableServices(metrics,
createMockTimeline(instants)).performTableServices(Option.empty(), true);
+
+ verify(metrics, times(1)).updateDeltaCommitsSinceLastCompaction(2L);
+ }
+
+ @Test
+ void performTableServicesReportsZeroDeltaCommitsOnEmptyTimeline() throws
Exception {
+ HoodieMetadataMetrics metrics = mock(HoodieMetadataMetrics.class);
+ writerForTableServices(metrics, createMockTimeline(new ArrayList<>()))
+ .performTableServices(Option.empty(), true);
+
+ // No completed delta commit yet, so the gauge is reset rather than left
at a stale value.
+ verify(metrics, times(1)).updateDeltaCommitsSinceLastCompaction(0L);
+ }
+
+ /**
+ * Builds a partially mocked writer whose {@code performTableServices} runs
for real against the given
+ * metadata timeline.
+ */
+ private static HoodieBackedTableMetadataWriter
writerForTableServices(HoodieMetadataMetrics metrics,
Review Comment:
Done. `testPerformTableServicesWithFailureHandling` now goes through the
same helper, with real timelines instead of deep stubs and `setField` instead
of the hand-rolled reflection.
--
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]
