voonhous commented on code in PR #18942:
URL: https://github.com/apache/hudi/pull/18942#discussion_r3829506189
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java:
##########
@@ -456,8 +459,12 @@ public void updateClusteringFileCreationMetrics(long
durationInMs) {
reportMetrics(HoodieTimeline.CLUSTERING_ACTION, "fileCreationTime",
durationInMs);
}
- public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline) {
- updateEarliestPendingInstant(activeTimeline,
EARLIEST_PENDING_CLUSTERING_INSTANT_STR, HoodieTimeline.CLUSTERING_ACTION);
+ public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline, final InstantGenerator instantGenerator) {
Review Comment:
Selection logic is fine, but this method now does storage IO on every commit
even when metrics are disabled -- and disabled is the default.
Before this PR `updateTableServiceInstantMetrics` was pure in-memory
action-name matching, so the missing guard cost nothing. Now the predicate
reaches `timeline.readRequestedReplaceMetadata(...)` ->
`readNonEmptyInstantContent` -> `ActiveTimelineV2.getContentStream` ->
`metaClient.getStorage().open(...)` for every pending replacecommit, inside
`postCommit`. `hoodie.metrics.on` defaults to `false`
(`HoodieMetricsConfig.java:58-62`), and the only guard is inside `updateMetric`
at `HoodieMetrics.java:583`, i.e. after the reads have already happened.
This is also the one entry point that `1fd2c3671a80` ("fix(metrics): NPE
handling when hudi metrics is disabled", #18947) missed:
`testUpdateMethodsAreNoOpsWhenMetricsOff` (`TestHoodieMetrics.java:411`)
enumerates 19 update methods and this is not one of them.
Please add the guard before merge, and add
`updateTableServiceInstantMetrics` to that test:
```suggestion
public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline, final InstantGenerator instantGenerator) {
if (!config.isMetricsOn()) {
return;
}
```
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java:
##########
@@ -484,13 +491,35 @@ private void updateEarliestPendingInstant(final
HoodieActiveTimeline activeTimel
final String metricName,
final String action) {
Set<String> validActions = CollectionUtils.createSet(action);
- HoodieTimeline filteredInstants =
activeTimeline.filterInflightsAndRequested().filter(instant ->
validActions.contains(instant.getAction()));
+ updateEarliestPendingInstant(activeTimeline, metricName, action, instant
-> validActions.contains(instant.getAction()));
+ }
+
+ private void updateEarliestPendingInstant(final HoodieActiveTimeline
activeTimeline,
+ final String metricName,
+ final String action,
+ final Predicate<HoodieInstant>
pendingFilter) {
+ HoodieTimeline filteredInstants =
activeTimeline.filterInflightsAndRequested().filter(pendingFilter);
Option<HoodieInstant> hoodieInstantOption =
filteredInstants.firstInstant();
if (hoodieInstantOption.isPresent()) {
updateTimestampMetric(metricName, action, hoodieInstantOption);
}
}
+ /**
+ * Whether a pending instant belongs to a clustering operation, on any table
version.
+ * A plan that cannot be read is treated as non-clustering so that metrics
never fail the commit.
+ */
+ private boolean isPendingClusteringInstant(final HoodieActiveTimeline
activeTimeline,
+ final HoodieInstant instant,
+ final InstantGenerator
instantGenerator) {
+ try {
+ return ClusteringUtils.isClusteringInstant(activeTimeline, instant,
instantGenerator);
+ } catch (Exception e) {
+ log.warn("Failed to read the replace metadata of {} while updating
clustering instant metrics", instant, e);
+ return false;
+ }
Review Comment:
The catch itself is right and load-bearing -- `postCommit`
(`BaseHoodieWriteClient.java:663-680`) has a `finally` but no `catch`, and
`hoodie.write.can.ignore.post.commit.failures` defaults to `false`
(`HoodieWriteConfig.java:747-748`), so an escaping exception would fail the
commit. That was impossible before this PR because the method did no IO. Worth
keeping.
Two follow-ups:
1. **Nothing exercises it.** Add a case whose timeline throws from the
metadata read, asserting `assertDoesNotThrow(...)` and
`pendingClusteringInstantCount == 0`. Without it the fallback degrades silently
to exactly the bug being fixed (count 0) and no test would notice.
2. **`warn` with a stacktrace is too loud for the case that will actually
hit this.** `ClusteringUtils.isClusteringInstant` calls the 3-arg
`getClusteringPlan`, which passes `Option.empty()` for the metaClient
(`ClusteringUtils.java:271-272`) and so deliberately opts out of the
concurrent-rollback recovery added by `f64c93ee899c` (#18288, "do not fail if
replacecommit was rolled back already (by a concurrent writer)"). A pending
replacecommit rolled back between timeline load and `postCommit` therefore logs
a full stacktrace once per instant per commit until the timeline is reloaded.
Please log at `debug`, or drop the exception argument.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java:
##########
@@ -456,8 +459,12 @@ public void updateClusteringFileCreationMetrics(long
durationInMs) {
reportMetrics(HoodieTimeline.CLUSTERING_ACTION, "fileCreationTime",
durationInMs);
}
- public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline) {
- updateEarliestPendingInstant(activeTimeline,
EARLIEST_PENDING_CLUSTERING_INSTANT_STR, HoodieTimeline.CLUSTERING_ACTION);
+ public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline, final InstantGenerator instantGenerator) {
+ // Clustering is scheduled as CLUSTERING_ACTION only on timeline layout 2.
On table version six it is scheduled as
+ // REPLACE_COMMIT_ACTION, which insert_overwrite and delete_partition
share, so the clustering plan is what
+ // identifies it rather than the action name.
Review Comment:
nit, feel free to ignore: this names the wrong field. Selection is on the
requested replace metadata's `operationType`, not on the clustering plan --
`ClusteringUtils.java:285` is
`WriteOperationType.CLUSTER.name().equals(requestedReplaceMetadata.get().getOperationType())`,
and the plan is only read once that check has passed. The test's own fixture
proves it: both instants carry a non-null plan and only the operationType
differs. Same wording is in the commit message and the PR body.
```suggestion
// Clustering is scheduled as CLUSTERING_ACTION only on timeline layout
2. On table version six it is scheduled as
// REPLACE_COMMIT_ACTION, which insert_overwrite and delete_partition
share, so the requested replace metadata's
// operationType is what identifies it rather than the action name.
```
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java:
##########
@@ -456,8 +459,12 @@ public void updateClusteringFileCreationMetrics(long
durationInMs) {
reportMetrics(HoodieTimeline.CLUSTERING_ACTION, "fileCreationTime",
durationInMs);
}
- public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline) {
- updateEarliestPendingInstant(activeTimeline,
EARLIEST_PENDING_CLUSTERING_INSTANT_STR, HoodieTimeline.CLUSTERING_ACTION);
+ public void updateTableServiceInstantMetrics(final HoodieActiveTimeline
activeTimeline, final InstantGenerator instantGenerator) {
+ // Clustering is scheduled as CLUSTERING_ACTION only on timeline layout 2.
On table version six it is scheduled as
+ // REPLACE_COMMIT_ACTION, which insert_overwrite and delete_partition
share, so the clustering plan is what
+ // identifies it rather than the action name.
+ Predicate<HoodieInstant> pendingClustering = instant ->
isPendingClusteringInstant(activeTimeline, instant, instantGenerator);
+ updateEarliestPendingInstant(activeTimeline,
EARLIEST_PENDING_CLUSTERING_INSTANT_STR, HoodieTimeline.CLUSTERING_ACTION,
pendingClustering);
Review Comment:
This predicate re-implements a timeline API that already exists and is
already layout-aware, and it costs twice the IO it needs to.
`HoodieTimeline.filterPendingClusteringTimeline()` (declared
`HoodieTimeline.java:441`) is implemented per layout:
- `BaseTimelineV1.java:62-68` -- `REPLACE_COMMIT_ACTION && !isCompleted() &&
ClusteringUtils.isClusteringInstant(this, i, instantGenerator)`, identical to
this predicate but using the timeline's own generator.
- `BaseTimelineV2.java:67-71` -- `CLUSTERING_ACTION && !isCompleted()`, zero
IO.
Two consequences of hand-rolling it here:
1. **The plan read is paid on every layout.** On tv8/9/10, the common case,
the layout already guarantees clustering has its own action, yet this now opens
and avro-parses `<ts>.replacecommit.requested` for every pending
insert_overwrite / delete_partition / bucket_rescale on every commit, only ever
to return false.
2. **It runs twice.** `filter()` is eager (`BaseHoodieTimeline.java:358` ->
`:96`, `setInstants(instants.collect(toList()))`), so `firstInstant()` at
`:502` does not short-circuit, and the same predicate object is filtered again
independently at `:578`. That is 2N `storage.open()` round trips per commit for
N pending replacecommits, and 4N for any instant whose requested file is
zero-length, since `ClusteringUtils.java:196-201` falls back to a second full
read via `isEmptyReplaceOrClusteringInstant`.
Concretely: drop the `Predicate` overloads, the `InstantGenerator` parameter
and both `BaseHoodieWriteClient` hunks, and materialize the timeline once:
```java
HoodieTimeline pendingClustering =
activeTimeline.filterPendingClusteringTimeline();
```
then feed `pendingClustering.firstInstant()` and
`pendingClustering.countInstants()` to the two clustering metrics. The PR
shrinks to a single production file.
One semantic difference to call out in the description if you take this: on
layout 2 it will not count a pending `replacecommit` clustering left over from
a v6 table. `UpgradeDowngrade` rolls back failed writes before upgrading so it
should not arise, but state it explicitly rather than paying the read
everywhere to cover it.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metrics/TestHoodieMetrics.java:
##########
@@ -320,13 +324,61 @@ public void testTimerCtxandGauges() throws
InterruptedException {
assertEquals((long)metrics.getRegistry().getGauges().get(metricName).getValue(),
6L);
}
+ @Test
+ void testPendingClusteringInstantMetricsOnTableVersionSix() {
+ // Table version six schedules clustering as REPLACE_COMMIT_ACTION, which
insert_overwrite shares.
+ HoodieInstant pendingInsertOverwrite =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1001");
+ HoodieInstant pendingClustering =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1002");
+ Map<String, String> operationTypes = new HashMap<>();
+ operationTypes.put("1001", WriteOperationType.INSERT_OVERWRITE.name());
+ operationTypes.put("1002", WriteOperationType.CLUSTER.name());
Review Comment:
Two gaps in this fixture, both on branches the fix depends on.
**No INFLIGHT instant.** `filterInflightsAndRequested()`
(`BaseHoodieTimeline.java:151-155`) returns REQUESTED *and* INFLIGHT; both
fixtures here are REQUESTED. The inflight case takes a different branch --
`ClusteringUtils.java:182-186` re-derives the requested instant from the
inflight one, with the comment "inflight replacecommit files don't have
clustering plan. This is because replacecommit inflight can have workload
profile for 'insert_overwrite'." That branch exists because of #2389
([HUDI-1498], "Read clustering plan from requested file for inflight instant"),
and on tv6 it is live, since inflight clustering is
`<ts>.replacecommit.inflight`. Add a third instant in `State.INFLIGHT` with a
CLUSTER operationType and assert the count becomes 2.
**The negative case is a shape production never writes.**
`"INSERT_OVERWRITE"` is not what lands on disk. Every replacecommit goes
through `createRequestedCommitWithReplaceMetadata`, which writes a bare `new
HoodieRequestedReplaceMetadata()` (`ActiveTimelineV1.java:138`,
`ActiveTimelineV2.java:143`) -- `operationType == null`, `clusteringPlan ==
null`, all fields being nullable in `HoodieRequestedReplaceMetadata.avsc`. So
the real-world negative case is a null operationType, and a future tightening
to `metadata.getOperationType().equals(CLUSTER.name())` would NPE on every
pending insert_overwrite, be swallowed by the new `catch`, and leave this test
green. Add an instant with a null operationType and assert the count is
unchanged. `DELETE_PARTITION` (`BaseHoodieTableServiceClient.java:789`) is the
one that genuinely writes a non-null operationType, and the description names
it, so it is worth a case too.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metrics/TestHoodieMetrics.java:
##########
@@ -320,13 +324,61 @@ public void testTimerCtxandGauges() throws
InterruptedException {
assertEquals((long)metrics.getRegistry().getGauges().get(metricName).getValue(),
6L);
}
+ @Test
+ void testPendingClusteringInstantMetricsOnTableVersionSix() {
Review Comment:
This test is named for table version six, but nothing in it is table version
six -- so the PR's headline claim has no coverage behind it.
`MockHoodieActiveTimeline extends ActiveTimelineV2`
(`TestHoodieMetrics.java:349`), whose `getTimelineLayoutVersion()` is hardcoded
to `LAYOUT_VERSION_2` (`BaseTimelineV2.java:61-64`), and `INSTANT_GENERATOR` is
`DefaultInstantGenerator extends InstantGeneratorV2`
(`HoodieTestUtils.java:110`, `DefaultInstantGenerator.java:23`). A real tv6
table is `LAYOUT_VERSION_1` (`HoodieTableVersion.java:54`) and gets
`ActiveTimelineV1` + `InstantGeneratorV1` (`TimelineLayout.java:87-93`). The
assertion runs an entirely tv8 stack.
The override of `readRequestedReplaceMetadata` at `:369` also stubs out the
only method in the chain that touches storage or the layout-specific serde, so
`CommitMetadataSerDeV1`, `readNonEmptyInstantContent` and the
empty-requested-file fallback are all bypassed. It has to be stubbed, in fact:
`MockHoodieActiveTimeline` calls the no-arg `ActiveTimelineV2()`, which leaves
`metaClient` null (`ActiveTimelineV2.java:88`), so any real read would NPE --
and that NPE would then be swallowed by the new `catch (Exception)` and turned
back into count 0.
There is a ready-made tv6 harness in this same module.
`TestBaseHoodieTableServiceClient` extends `HoodieCommonTestHarness`, where
`initMetaClient(preTableVersion8)` yields `HoodieTableVersion.SIX`
(`HoodieCommonTestHarness.java:191-197`), and
`createClusteringInstant(instantTime, REPLACE_COMMIT_ACTION)` at
`TestBaseHoodieTableServiceClient.java:429-449` writes a genuine pending
replacecommit clustering instant via `saveToPendingClusterCommit` and
transitions it to inflight.
`isClusteringInstantEligibleForRollback_returnsTrueWhenEligible` at `:299-315`
is the exact pattern to copy:
```java
@ParameterizedTest
@ValueSource(booleans = {true, false})
void pendingClusteringInstantMetrics(boolean preTableVersion8) throws
IOException {
initMetaClient(preTableVersion8);
String clusteringAction = preTableVersion8 ? REPLACE_COMMIT_ACTION :
CLUSTERING_ACTION;
...
}
```
Please rebuild the test on that harness so both layouts run against real
instant files. If you would rather keep the mock, rename this to something
honest such as `testPendingClusteringMetricsSelectByOperationType`, so it does
not claim tv6 coverage it does not have.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metrics/TestHoodieMetrics.java:
##########
@@ -320,13 +324,61 @@ public void testTimerCtxandGauges() throws
InterruptedException {
assertEquals((long)metrics.getRegistry().getGauges().get(metricName).getValue(),
6L);
}
+ @Test
+ void testPendingClusteringInstantMetricsOnTableVersionSix() {
+ // Table version six schedules clustering as REPLACE_COMMIT_ACTION, which
insert_overwrite shares.
+ HoodieInstant pendingInsertOverwrite =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1001");
+ HoodieInstant pendingClustering =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1002");
+ Map<String, String> operationTypes = new HashMap<>();
+ operationTypes.put("1001", WriteOperationType.INSERT_OVERWRITE.name());
+ operationTypes.put("1002", WriteOperationType.CLUSTER.name());
+
+ hoodieMetrics.updateTableServiceInstantMetrics(
+ new MockClusteringPlanTimeline(operationTypes, pendingInsertOverwrite,
pendingClustering), INSTANT_GENERATOR);
+
+ // Only the instant carrying a clustering plan counts. insert_overwrite is
the earlier of the two, so
+ // both assertions fail if the action name alone is used to select it.
Review Comment:
nit, feel free to ignore: same wrong field as the production comment. Both
fixtures carry a non-null clustering plan; what separates them is
`operationType` (`ClusteringUtils.java:285`). The non-null plan on the
insert_overwrite instant is actually the valuable part of this fixture -- it
pins that selection must not key on plan presence -- so it is worth saying that
rather than describing it backwards.
```suggestion
// Only the instant whose requested replace metadata carries
operationType CLUSTER counts. Both fixtures
// deliberately carry a plan, so this also pins that selection must not
key on plan presence.
// insert_overwrite is the earlier of the two, so both assertions fail
if the action name alone selects it.
```
Separately, both assertions below dereference `...getGauges().get(name)`
with no null check. If the fix regresses, `updateEarliestPendingInstant` never
registers the gauge (`HoodieMetrics.java:501-505` writes only when an instant
matches), so line 346 fails with an NPE rather than a readable assertion. An
`assertNotNull(gauge, ...)` first would make that failure diagnosable.
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metrics/TestHoodieMetrics.java:
##########
@@ -320,13 +324,61 @@ public void testTimerCtxandGauges() throws
InterruptedException {
assertEquals((long)metrics.getRegistry().getGauges().get(metricName).getValue(),
6L);
}
+ @Test
+ void testPendingClusteringInstantMetricsOnTableVersionSix() {
+ // Table version six schedules clustering as REPLACE_COMMIT_ACTION, which
insert_overwrite shares.
+ HoodieInstant pendingInsertOverwrite =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1001");
+ HoodieInstant pendingClustering =
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "1002");
+ Map<String, String> operationTypes = new HashMap<>();
+ operationTypes.put("1001", WriteOperationType.INSERT_OVERWRITE.name());
+ operationTypes.put("1002", WriteOperationType.CLUSTER.name());
+
+ hoodieMetrics.updateTableServiceInstantMetrics(
+ new MockClusteringPlanTimeline(operationTypes, pendingInsertOverwrite,
pendingClustering), INSTANT_GENERATOR);
+
+ // Only the instant carrying a clustering plan counts. insert_overwrite is
the earlier of the two, so
+ // both assertions fail if the action name alone is used to select it.
+ String countMetric =
hoodieMetrics.getMetricsName(HoodieTimeline.CLUSTERING_ACTION,
HoodieMetrics.PENDING_CLUSTERING_INSTANT_COUNT_STR);
+ assertEquals(1L, (long)
metrics.getRegistry().getGauges().get(countMetric).getValue());
+ String earliestMetric =
hoodieMetrics.getMetricsName(HoodieTimeline.CLUSTERING_ACTION,
HoodieMetrics.EARLIEST_PENDING_CLUSTERING_INSTANT_STR);
+ assertEquals(1002L, (long)
metrics.getRegistry().getGauges().get(earliestMetric).getValue());
+ }
+
private static class MockHoodieActiveTimeline extends ActiveTimelineV2 {
public MockHoodieActiveTimeline(HoodieInstant... instants) {
super();
this.setInstants(Arrays.asList(instants));
}
}
+ /**
+ * Serves a requested replace metadata per instant time so pending
replacecommits can be told apart
+ * by their write operation type without a backing timeline on storage.
+ */
+ private static class MockClusteringPlanTimeline extends
MockHoodieActiveTimeline {
Review Comment:
nit, feel free to ignore: this `HoodieRequestedReplaceMetadata` builder
chain is byte-for-byte identical to
`TestBaseHoodieTableServiceClient.java:434-441` in this same module. If the
test moves onto that harness it disappears entirely; if the mock stays,
consider factoring the metadata literal into a shared test helper rather than
keeping a third copy.
--
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]
