voonhous commented on code in PR #19575:
URL: https://github.com/apache/hudi/pull/19575#discussion_r3853268963
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java:
##########
@@ -255,6 +256,12 @@ public boolean commitStats(String instantTime,
TableWriteStats tableWriteStats,
return true;
}
extraMetadata = updateExtraMetadata(extraMetadata);
+ // The commit is the boundary for the record index lookup counters.
Snapshot them into this commit's
+ // metadata now, but only release them from the registry once the commit
has actually landed --
+ // hooked here rather than in updateExtraMetadata because that is shared
with table-service
+ // scheduling, which would otherwise consume the counters into a
compaction or clustering plan.
+ ExecutorMetrics.DrainedCounters executorCounters =
+ ExecutorMetrics.snapshotIntoCommitMetadata(extraMetadata.get(),
config);
Review Comment:
`snapshotIntoCommitMetadata` mutates the enriched `extraMetadata` map, and
the same `Option` goes to `runTableServicesInline` at L302. From there
`scheduleTableServiceInternal` re-enriches with `putAll`,
`ScheduleCompactionActionExecutor:109` puts it on the plan, and
`CompactHelpers:85` copies the plan's extra metadata into the completed
compaction commit, so an inline compaction carries the triggering commit's
`hoodie.rli.lookup.*` values. That is the misattribution the comment above says
this placement avoids, and no new test enables inline services. Could we stamp
into a copy used only for `CommitUtils.buildMetadata`, and add one MOR
inline-compaction case asserting the compaction commit has no
`hoodie.rli.lookup.*` key?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/ExecutorMetrics.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.metrics;
+
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Commit-boundary drain for executor-collected metrics, generic over {@link
ExecutorMetricRegistry}. On
+ * the shared commit path, so it covers Spark DataSource, Spark SQL and
DeltaStreamer alike.
+ */
+public class ExecutorMetrics {
+
+ private ExecutorMetrics() {
+ }
+
+ /**
+ * Snapshots into commit metadata without consuming. Split from {@link
#publishAndRelease} so a commit
+ * that never lands neither loses its counters nor publishes gauges for
rolled-back work. An all-zero
+ * registry is skipped to keep residue off the timeline; zeros are otherwise
kept, since an explicit
+ * {@code misses=0} is meaningful.
+ */
+ public static DrainedCounters snapshotIntoCommitMetadata(Map<String, String>
commitMetadata,
+ HoodieWriteConfig
config) {
+ return snapshotIntoCommitMetadata(commitMetadata, config,
Arrays.asList(ExecutorMetricRegistry.values()));
+ }
+
+ /** Visible for testing the collection machinery against a group it does not
ship with. */
+ static DrainedCounters snapshotIntoCommitMetadata(Map<String, String>
commitMetadata,
+ HoodieWriteConfig config,
+ Collection<? extends
ExecutorMetricGroup> groups) {
+ List<Drained> drained = new ArrayList<>();
+ for (ExecutorMetricGroup metricRegistry : groups) {
+ if (!metricRegistry.isEnabled(config)) {
+ continue;
+ }
+ Registry registry = Registry.REGISTRY_MAP.get(
+ Registry.makeKey(config.getTableName(),
metricRegistry.scopedName(config.getBasePath())));
+ if (registry == null) {
+ continue;
+ }
+ Map<String, Long> counts = new HashMap<>();
+ boolean recordedSomething = false;
+ for (Map.Entry<String, Long> counter :
registry.getAllCounts(false).entrySet()) {
+ if (counter.getValue() == null) {
+ continue;
+ }
+ counts.put(counter.getKey(), counter.getValue());
+ recordedSomething |= counter.getValue() != 0L;
+ }
+ if (!recordedSomething) {
Review Comment:
A commit that performs no lookup (insert-only, bulk_insert, an empty
micro-batch) takes this `continue`, so the `rli.lookup.*` gauges registered by
an earlier commit keep that commit's values and a scheduled reporter
(pushgateway) re-emits them indefinitely; per key, `dedupe.*` goes stale while
`tag.*` updates, so the two sinks disagree. HUDI-3373 (#4760) added
`HoodieMetrics.updateMetricsForEmptyData` for exactly this on the `commit.*`
gauges. Could we publish explicit zeros for the group's counter names when the
registry is missing or all-zero (keeping the timeline skip), or add these names
to `updateMetricsForEmptyData`?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/ExecutorMetrics.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.metrics;
+
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Commit-boundary drain for executor-collected metrics, generic over {@link
ExecutorMetricRegistry}. On
+ * the shared commit path, so it covers Spark DataSource, Spark SQL and
DeltaStreamer alike.
+ */
+public class ExecutorMetrics {
+
+ private ExecutorMetrics() {
+ }
+
+ /**
+ * Snapshots into commit metadata without consuming. Split from {@link
#publishAndRelease} so a commit
+ * that never lands neither loses its counters nor publishes gauges for
rolled-back work. An all-zero
+ * registry is skipped to keep residue off the timeline; zeros are otherwise
kept, since an explicit
+ * {@code misses=0} is meaningful.
+ */
+ public static DrainedCounters snapshotIntoCommitMetadata(Map<String, String>
commitMetadata,
+ HoodieWriteConfig
config) {
+ return snapshotIntoCommitMetadata(commitMetadata, config,
Arrays.asList(ExecutorMetricRegistry.values()));
+ }
+
+ /** Visible for testing the collection machinery against a group it does not
ship with. */
+ static DrainedCounters snapshotIntoCommitMetadata(Map<String, String>
commitMetadata,
+ HoodieWriteConfig config,
+ Collection<? extends
ExecutorMetricGroup> groups) {
+ List<Drained> drained = new ArrayList<>();
+ for (ExecutorMetricGroup metricRegistry : groups) {
+ if (!metricRegistry.isEnabled(config)) {
+ continue;
+ }
+ Registry registry = Registry.REGISTRY_MAP.get(
+ Registry.makeKey(config.getTableName(),
metricRegistry.scopedName(config.getBasePath())));
+ if (registry == null) {
+ continue;
+ }
+ Map<String, Long> counts = new HashMap<>();
+ boolean recordedSomething = false;
+ for (Map.Entry<String, Long> counter :
registry.getAllCounts(false).entrySet()) {
+ if (counter.getValue() == null) {
+ continue;
+ }
+ counts.put(counter.getKey(), counter.getValue());
+ recordedSomething |= counter.getValue() != 0L;
+ }
+ if (!recordedSomething) {
+ continue;
+ }
+ counts.forEach((name, value) ->
+ commitMetadata.put(metricRegistry.commitMetadataPrefix() + name,
String.valueOf(value)));
+ drained.add(new Drained(metricRegistry, registry, counts));
+ }
+ return drained.isEmpty() ? DrainedCounters.EMPTY : new
DrainedCounters(drained);
+ }
+
+ /**
+ * Release subtracts what was published rather than clearing, so a straggler
task's update arriving after
+ * the snapshot survives. Publishing here rather than letting the reporter
scrape is what lets both sinks
+ * work at once: {@link Registry#getAllMetrics} consumes the registry when
it scrapes.
+ */
+ public static void publishAndRelease(DrainedCounters counters, HoodieMetrics
hoodieMetrics) {
Review Comment:
A failed or abandoned attempt's counters are stamped on whatever commits
next, so that commit's `key_count` no longer describes its own records. Your
earlier branch fixed exactly this at `dd4a48c` (clear at `startCommit`: "an
abandoned 80-key attempt followed by a 10-key commit reported 80"); this PR
reverses it and asserts the opposite. Could we restore clear-at-`startCommit`,
flip the `AcrossFailedCommit` assertion to "the retry reports only its own
lookups", and add an insert-after-failed-upsert case?
<details><summary>Why clear-at-startCommit is safe, and what else picks up
leftovers</summary>
- `startCommit` precedes every in-tree lookup: `HoodieSparkSqlWriter:533`
before `:553 handleInsertDuplicates`; `StreamSync:877` before `:1104`;
write-path `tagLocation` runs inside `upsert/delete`.
- Leftovers also land on `commitTableChange` (ALTER SCHEMA,
`BaseHoodieWriteClient:1872`) and on a successful empty commit
(`hoodie.allow.empty.commit=false` returns at L255, above the snapshot).
- With `hoodie.metrics.on=true` the carry-over does not hold anyway:
`DefaultSource:194 cleanup()` -> `Metrics.shutdownAllMetrics()` ->
`Registry.getAllMetrics(true, true)` clears every registry after each
DataSource write.
</details>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDReadClient.java:
##########
@@ -210,8 +212,16 @@ public JavaRDD<HoodieRecord<T>>
filterExists(JavaRDD<HoodieRecord<T>> hoodieReco
* @return Tagged RDD of Hoodie records
*/
public JavaRDD<HoodieRecord<T>> tagLocation(JavaRDD<HoodieRecord<T>>
hoodieRecords) throws HoodieIndexException {
- return HoodieJavaRDD.getJavaRDD(
- index.tagLocation(HoodieJavaRDD.of(hoodieRecords), context,
hoodieTable));
+ // Lookups driven from the read client are dedupe traffic, not
tag-location traffic. Label them so
+ // the two are distinguishable in the reported counters. Driver-side only:
the label is
+ // captured when the lookup closure is built.
+ String previousCaller =
RecordIndexLookupMetrics.setCaller(RecordIndexMetricNames.CALLER_DEDUPE);
Review Comment:
`CALLER_DEDUPE` has no test coverage: `grep -rn CALLER_DEDUPE` under
`src/test` returns nothing, and `TestRliMetricsOnStreamerPath` asserts only
`tag.*`. It is reachable on the DataSource path too (`insert.drop.duplicates`
-> `HoodieSparkSqlWriter:553` -> `resolveDuplicates`), which is cheaper to test
than a streamer arm. Also `checkExists` (L188) calls `index.tagLocation`
directly, so it is labelled `tag` while `filterExists` is `dedupe`, contrary to
the comment above. Could we add one DataSource case with
`insert.drop.duplicates=true` asserting both `dedupe.*` and `tag.*` on the same
commit, and set the label once for the whole read client?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/RecordIndexLookupMetrics.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.index;
+
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.metrics.DistributedRegistry;
+import org.apache.hudi.metrics.ExecutorMetricGroup;
+import org.apache.hudi.metrics.ExecutorMetricRegistry;
+import org.apache.hudi.metrics.RecordIndexMetricNames;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/** Executor-side emission for the record index lookup counters. */
+public class RecordIndexLookupMetrics {
+
+ /** Set by the read client around its own tagging call, so dedupe traffic is
attributable separately. */
+ private static final ThreadLocal<String> CALLER =
+ ThreadLocal.withInitial(() ->
RecordIndexMetricNames.CALLER_TAG_LOCATION);
+
+ private RecordIndexLookupMetrics() {
+ }
+
+ public static String currentCaller() {
+ return CALLER.get();
+ }
+
+ /** Restore rather than clear, so a nested tagging call does not reset the
label. */
+ public static String setCaller(String caller) {
+ String previous = CALLER.get();
+ CALLER.set(caller);
+ return previous;
+ }
+
+ public static void restoreCaller(String previous) {
+ CALLER.set(previous);
+ }
+
+ /**
+ * The registries a lookup task collects into, keyed by bare name. Includes
every entry on
+ * {@link ExecutorMetricRegistry}. Delivery is by closure capture, which is
deterministic; resolution is
+ * by name, which lets code below the write API take part without a
signature change.
+ */
+ public static Map<String, Registry> resolveBundle(HoodieEngineContext
context, HoodieWriteConfig config) {
+ return resolveBundle(context, config,
Arrays.asList(ExecutorMetricRegistry.values()));
+ }
+
+ /** Visible for testing the bundle against a group the enum does not ship
with. */
+ public static Map<String, Registry> resolveBundle(HoodieEngineContext
context, HoodieWriteConfig config,
+ Collection<? extends
ExecutorMetricGroup> groups) {
+ Map<String, Registry> bundle = new HashMap<>();
+ for (ExecutorMetricGroup metricRegistry : groups) {
+ if (!metricRegistry.isEnabled(config)) {
+ continue;
+ }
+ Registry registry = context.getMetricRegistry(config.getTableName(),
Review Comment:
`getMetricRegistry` starts with `tableName.isEmpty()`, and
`config.getTableName()` can be null: `TBL_NAME` has no default and
`Builder.validate()` only requires `BASE_PATH`. Pre-PR that line sat behind
`hoodie.metrics.on`; now every RLI `tagLocation` reaches it, so the public
`SparkRDDReadClient(ctx, basePath, sqlCtx, RECORD_INDEX)` ctor (L104) or any
write config built without `forTable()` NPEs on an RLI table. Could we skip
collection when the table name is null or empty?
```suggestion
String tableName = config.getTableName();
if (tableName == null || tableName.isEmpty()) {
continue;
}
Registry registry = context.getMetricRegistry(tableName,
```
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/ExecutorMetricRegistry.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.metrics;
+
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.function.Predicate;
+
+/**
+ * Every class of executor-collected metric, and the only thing a new one is
added to. The driver must
+ * declare it up front because an {@code AccumulatorV2} must be registered
with the {@code SparkContext}
+ * before a task can contribute; the bundle sent to executors and the commit
drain both iterate this.
+ */
+public enum ExecutorMetricRegistry implements ExecutorMetricGroup {
+
+ RECORD_INDEX_LOOKUP(
+ "HoodieRecordIndexLookup",
+ "hoodie.rli.lookup.",
Review Comment:
Once written these keys are permanent timeline content, so the naming is
worth settling now. No existing extra-metadata key starts with `hoodie.`
(`schema`, `deltastreamer.checkpoint.key`, `latest_schema`, and #18183's
`hudi.version` / `engine` / `config.<key>`, which deliberately prefixes real
config keys with `config.`), so `hoodie.` reads as a config key that does not
exist. The full key also repeats itself:
`hoodie.rli.lookup.tag.lookup_record_index_key_count`. Could the prefix be
`rli.lookup.` (or `metrics.record.index.lookup.`) and the counters drop the
`lookup_record_index_` segment, e.g. `rli.lookup.tag.key_count`?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestRliMetricsOnStreamerPath.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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.deltastreamer;
+
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.metrics.RecordIndexMetricNames;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The RLI lookup counters must reach commit metadata on the DeltaStreamer
path, not only on the Spark DataSource path.
+ */
+@Tag("functional")
+public class TestRliMetricsOnStreamerPath extends HoodieDeltaStreamerTestBase {
+
+ /** Selects the global or partitioned record index. */
+ private static void enableRecordIndex(HoodieDeltaStreamer.Config cfg,
boolean partitioned) {
+ cfg.configs.add(HoodieMetadataConfig.ENABLE.key() + "=true");
+
cfg.configs.add(HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key()
+ "=" + !partitioned);
+ cfg.configs.add(HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key()
+ "=" + partitioned);
+ cfg.configs.add(HoodieIndexConfig.INDEX_TYPE.key() + "="
+ + (partitioned ? "RECORD_LEVEL_INDEX" : "GLOBAL_RECORD_LEVEL_INDEX"));
+ }
+
+ private static Map<String, String> rliCountersOnLatestCommit(String
tableBasePath) throws Exception {
+ HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder()
+ .setConf(HoodieTestUtils.getDefaultStorageConf())
+ .setBasePath(tableBasePath)
+ .build();
+ metaClient.reloadActiveTimeline();
+ HoodieInstant lastInstant = metaClient.getActiveTimeline()
+ .getCommitsTimeline().filterCompletedInstants().lastInstant().get();
+ HoodieCommitMetadata commitMetadata =
metaClient.getActiveTimeline().readCommitMetadata(lastInstant);
+ Map<String, String> rli = new HashMap<>();
+ commitMetadata.getExtraMetadata().forEach((k, v) -> {
+ if (k.startsWith(RecordIndexMetricNames.COMMIT_METADATA_PREFIX)) {
+ rli.put(k, v);
+ }
+ });
+ return rli;
+ }
+
+ private static String tagKey(String metric) {
+ return RecordIndexMetricNames.COMMIT_METADATA_PREFIX
+ +
RecordIndexMetricNames.key(RecordIndexMetricNames.CALLER_TAG_LOCATION, metric);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testRliCountersReachCommitMetadataOnStreamerPath(boolean
partitioned) throws Exception {
+ String label = partitioned ? "partitioned" : "global";
+ String tableBasePath = basePath + "/test_rli_metrics_streamer_" + label;
+
+ // Sync 1 -- build the table and the record index.
+ HoodieDeltaStreamer.Config insertCfg =
+ TestHoodieDeltaStreamer.TestHelpers.makeConfig(tableBasePath,
WriteOperationType.INSERT);
+ enableRecordIndex(insertCfg, partitioned);
+ new HoodieDeltaStreamer(insertCfg, jsc).sync();
+
+ // Sync 2 -- upsert, which tags incoming keys against the record index.
+ HoodieDeltaStreamer.Config upsertCfg =
+ TestHoodieDeltaStreamer.TestHelpers.makeConfig(tableBasePath,
WriteOperationType.UPSERT);
+ enableRecordIndex(upsertCfg, partitioned);
+ new HoodieDeltaStreamer(upsertCfg, jsc).sync();
+
+ Map<String, String> counters = rliCountersOnLatestCommit(tableBasePath);
+
+ System.out.println("\n===== DeltaStreamer (" + label + " RLI) -- RLI
counters on the commit =====");
+ if (counters.isEmpty()) {
+ System.out.println(" (none found)");
+ } else {
+ counters.entrySet().stream()
+ .sorted(Map.Entry.comparingByKey())
+ .forEach(e -> System.out.println(String.format(" %-52s %s",
e.getKey(), e.getValue())));
+ }
+
System.out.println("==========================================================\n");
+
+ assertFalse(counters.isEmpty(),
+ "the commit-boundary drain must fire on the DeltaStreamer path;
hudi-utilities never calls "
+ + "Metrics.shutdownAllMetrics, so nothing else would publish
these");
+
+ String lookedUp = tagKey(RecordIndexMetricNames.KEY_COUNT);
+ assertTrue(counters.containsKey(lookedUp),
+ "tag-location traffic must be attributed on the streamer path too; got
" + counters.keySet());
+
+ long records = Long.parseLong(counters.get(lookedUp));
+ long hits =
Long.parseLong(counters.get(tagKey(RecordIndexMetricNames.KEY_HIT_COUNT)));
+ long misses =
Long.parseLong(counters.get(tagKey(RecordIndexMetricNames.KEY_MISS_COUNT)));
+ assertTrue(records > 0, "the upsert sync looked up at least one key");
+ assertEquals(records, hits + misses, "hits + misses must account for every
key looked up");
Review Comment:
`records > 0` plus `records == hits + misses` cannot fail on a doubled
count: `misses` is computed as `records - hits` at the emission site
(`RecordIndexLookupMetrics:111`), so the identity holds by construction. This
is the only DeltaStreamer-path test and the values are deterministic (CI logged
`1000 / 500 / 500` on both arms: `sourceLimit=1000`, then 500 updates + 500
inserts). Could we assert the exact counts instead?
```suggestion
assertEquals(1000L, records, "the upsert sync looked up every key from
the first sync");
assertEquals(500L, hits, "the 500 updates hit the index");
assertEquals(500L, misses, "the 500 fresh inserts missed");
```
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/RecordIndexLookupMetrics.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.index;
+
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.metrics.DistributedRegistry;
+import org.apache.hudi.metrics.ExecutorMetricGroup;
+import org.apache.hudi.metrics.ExecutorMetricRegistry;
+import org.apache.hudi.metrics.RecordIndexMetricNames;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/** Executor-side emission for the record index lookup counters. */
+public class RecordIndexLookupMetrics {
+
+ /** Set by the read client around its own tagging call, so dedupe traffic is
attributable separately. */
+ private static final ThreadLocal<String> CALLER =
+ ThreadLocal.withInitial(() ->
RecordIndexMetricNames.CALLER_TAG_LOCATION);
+
+ private RecordIndexLookupMetrics() {
+ }
+
+ public static String currentCaller() {
+ return CALLER.get();
+ }
+
+ /** Restore rather than clear, so a nested tagging call does not reset the
label. */
+ public static String setCaller(String caller) {
+ String previous = CALLER.get();
+ CALLER.set(caller);
+ return previous;
+ }
+
+ public static void restoreCaller(String previous) {
+ CALLER.set(previous);
+ }
+
+ /**
+ * The registries a lookup task collects into, keyed by bare name. Includes
every entry on
+ * {@link ExecutorMetricRegistry}. Delivery is by closure capture, which is
deterministic; resolution is
+ * by name, which lets code below the write API take part without a
signature change.
+ */
+ public static Map<String, Registry> resolveBundle(HoodieEngineContext
context, HoodieWriteConfig config) {
+ return resolveBundle(context, config,
Arrays.asList(ExecutorMetricRegistry.values()));
+ }
+
+ /** Visible for testing the bundle against a group the enum does not ship
with. */
+ public static Map<String, Registry> resolveBundle(HoodieEngineContext
context, HoodieWriteConfig config,
+ Collection<? extends
ExecutorMetricGroup> groups) {
+ Map<String, Registry> bundle = new HashMap<>();
+ for (ExecutorMetricGroup metricRegistry : groups) {
+ if (!metricRegistry.isEnabled(config)) {
Review Comment:
This gate makes the RLI registry the first thing to enter
`DISTRIBUTED_REGISTRY_MAP` without opt-in (pre-PR: only
`DistributedRegistryUtil`, behind `hoodie.metrics.on` AND
`hoodie.metrics.executor.enable`). That static map is captured into every
`HoodieSparkEngineContext.map/mapToPair/flatMap` closure (L141-206), so every
such job for every table in the JVM now carries one accumulator per RLI table
per task, and nothing evicts. Could the gate follow `LOCK_METRICS_ENABLE`
(`.withInferFunction(cfg -> cfg.getBoolean(TURN_METRICS_ON))`), and could this
registry stay out of the engine-context captures, which only need the FS
registries?
<details><summary>Per-task cost</summary>
The closure carries an empty `copyAndReset` copy (cheap), but per task Spark
registers it with the `TaskContext`, ships an update back even when zero, and
retains one unnamed `AccumulableInfo` per task per accumulator in the driver
AppStatusStore/UI and the event log. The single-table overhead run in the
description cannot show this.
</details>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/DistributedRegistry.java:
##########
@@ -68,9 +79,36 @@ public void add(String name, long value) {
@Override
public void set(String name, long value) {
+ // Last-writer-wins is neither commutative nor associative, and the driver
merges executor copies in
+ // an unspecified order. Driver only; executors use increment()/add().
+ if (TaskContext.get() != null) {
Review Comment:
These guards throw from inside a task, and `release()`'s copy runs at
`BaseHoodieWriteClient:287` after the commit has landed, inside a `try` that
catches only `IOException`, so a metrics problem would surface as a failed
write on a completed instant. That contradicts the policy stated in
`NoOpRegistry` ("a metrics gap must not fail a write"), and
`Metrics.registerGauge` already swallows for the same reason. Could both guards
`LOG.warn` and return instead, and the drain at L264/L287 be wrapped in a
catch-all?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/RecordIndexMetricNames.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.metrics;
+
+/**
+ * Counter names for the record index lookup phase. Collection itself is
generic -- see
+ * {@link ExecutorMetricRegistry} and {@link ExecutorMetrics}.
+ */
+public class RecordIndexMetricNames {
+
+ /** Scoping, prefix and reporter naming all live on the enum entry. */
+ public static final String REGISTRY_NAME =
ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.registryName();
+
+ public static final String COMMIT_METADATA_PREFIX =
+ ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.commitMetadataPrefix();
+
+ // Counters are tagged by caller so dedupe traffic is distinguishable from
tag-location traffic.
+ public static final String CALLER_TAG_LOCATION = "tag";
+ public static final String CALLER_DEDUPE = "dedupe";
+
+ public static final String KEY_COUNT = "lookup_record_index_key_count";
Review Comment:
Two small things on the names. (1) The value counts records, not distinct
keys (`keysToLookup` has one entry per `HoodieRecord`), while the name says
`key_count` and the config doc says "records looked up". (2) L37-43 re-declare
the literal strings of `HoodieMetadataMetrics.LOOKUP_RECORD_INDEX_*`, dead
since `2a0d2aeef6a0` (HUDI-7391), and `HoodieBackedTableMetadata:356` still
carries the `TODO [HUDI-9544]` the description says this closes. Could the
counter be `record_count` (or the doc say keys), and could the dead constants
be referenced or deleted and the TODO updated?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/RecordIndexMetricNames.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.metrics;
+
+/**
+ * Counter names for the record index lookup phase. Collection itself is
generic -- see
+ * {@link ExecutorMetricRegistry} and {@link ExecutorMetrics}.
+ */
+public class RecordIndexMetricNames {
+
+ /** Scoping, prefix and reporter naming all live on the enum entry. */
+ public static final String REGISTRY_NAME =
ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.registryName();
+
+ public static final String COMMIT_METADATA_PREFIX =
+ ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.commitMetadataPrefix();
+
+ // Counters are tagged by caller so dedupe traffic is distinguishable from
tag-location traffic.
+ public static final String CALLER_TAG_LOCATION = "tag";
+ public static final String CALLER_DEDUPE = "dedupe";
+
+ public static final String KEY_COUNT = "lookup_record_index_key_count";
+ public static final String KEY_HIT_COUNT =
"lookup_record_index_key_hit_count";
+ public static final String KEY_MISS_COUNT =
"lookup_record_index_key_miss_count";
+ public static final String SHARDS_READ = "lookup_record_index_shards_read";
+ /** Wall-clock spent in the shard read, summed across shards. Revives the
third dead upstream constant,
+ * {@code HoodieMetadataMetrics.LOOKUP_RECORD_INDEX_TIME_STR}. */
+ public static final String LOOKUP_TIME = "lookup_record_index_time";
Review Comment:
`SparkRDDWriteClient:189` already publishes `index.lookup.duration` via
`HoodieMetrics.updateIndexMetrics` (driver wall-clock for the whole
`tagLocation`), so this adds a second lookup-time metric with a different
meaning (sum of per-shard executor time). Could we drop `LOOKUP_TIME`, or state
in the javadoc and the config doc how the two differ?
##########
hudi-common/src/main/java/org/apache/hudi/common/config/metrics/HoodieMetricsConfig.java:
##########
@@ -109,6 +109,17 @@ public class HoodieMetricsConfig extends HoodieConfig {
.sinceVersion("0.13.0")
.withDocumentation("Enable metrics for locking infra. Useful when
operating in multiwriter mode");
+ public static final ConfigProperty<Boolean> RLI_LOOKUP_METRICS_ENABLE =
ConfigProperty
Review Comment:
nit, feel free to ignore: every sibling boolean here has a `Builder` setter
(`withExecutorMetrics`, `withLockingMetrics`, `withCompactionLogBlockMetrics`).
Could this one get `withRecordIndexLookupMetrics(boolean)` for parity?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/HoodieSparkEngineContext.java:
##########
@@ -278,11 +279,43 @@ public String getApplicationId() {
return javaSparkContext.sc().applicationId();
}
+ /**
+ * Drops a registry from both process-wide maps. Only for tests that create
their own SparkContexts:
+ * without it they leave accumulators bound to stopped contexts behind for
whatever runs next in the
+ * same JVM.
+ */
+ @VisibleForTesting
+ public static void removeMetricRegistryForTesting(String tableName, String
registryName) {
+ DISTRIBUTED_REGISTRY_MAP.remove(tableName.isEmpty() ? registryName :
tableName + "." + registryName);
+ Registry.REGISTRY_MAP.remove(Registry.makeKey(tableName, registryName));
+ }
+
@Override
public Registry getMetricRegistry(String tableName, String registryName) {
Review Comment:
The process-wide map is what forces the SHA-256 digest, the stale-context
branch, the replacement branch below and `removeMetricRegistryForTesting`, and
it is the source of the null-`tableName` NPE and the unbounded growth raised in
`RecordIndexLookupMetrics`. #19063 lists "tying registry lifetime to
`HoodieEngineContext` / write client rather than the JVM process" as the
intended follow-up, and it looks reachable: `StreamSync:1104` reuses the write
client's context for dedupe, and `HoodieSparkSqlWriter:553` has
`client.getEngineContext` in scope (`DataSourceUtils:298` currently builds a
throwaway one). Could the registry be an instance field on
`HoodieSparkEngineContext` keyed by base path, with `commitStats` reading it
off `context`?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/DistributedRegistry.java:
##########
@@ -48,9 +51,17 @@ public String getName() {
public void register(JavaSparkContext jsc) {
if (!isRegistered()) {
jsc.sc().register(this);
+ // Only when this call actually registers: stamping unconditionally
would re-brand an accumulator
+ // bound to a dead context and mask the staleness this field exists to
detect.
+ this.registeredAppId = jsc.sc().applicationId();
}
}
+ /** False when bound to a different (typically stopped) context, meaning it
must be recreated. */
+ public boolean isRegisteredWith(JavaSparkContext jsc) {
Review Comment:
`isRegisteredWith` + the `compute` rewrite, the `set()` executor guard and
release-at-commit are gaps 2, 7 and 8 of #19063 (the guard is proposed there
verbatim), and they fix `hoodie.metrics.executor.enable` on master today for
the existing `HoodieWrapperFileSystem` user. That issue's own breakdown puts
them in PR 1 / PR 2, and the description here does not reference it. Could the
`DistributedRegistry` / `getMetricRegistry` hardening go out as its own PR
under #19063, leaving this one to add the counters on top?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/RliLookupMetricsTestBase.scala:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.functional
+
+import org.apache.hudi.DataSourceWriteOptions
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.metrics.Registry
+import org.apache.hudi.config.HoodieIndexConfig
+import org.apache.hudi.metrics.RecordIndexMetricNames
+
+import scala.collection.JavaConverters._
+
+/**
+ * Shared plumbing for the record level index lookup metric tests: index
selection, and reading the counters back the way an operator would -- off the la
+ */
+abstract class RliLookupMetricsTestBase extends RecordLevelIndexTestBase {
Review Comment:
The drain has one engine-agnostic hook (`commitStats:264/287`), but the
functional suite re-covers it 30 times: 79 full MDT+RLI table writes and 30
SparkContext restarts on the single UT_FT_2 job. Could we consolidate to
roughly 4 classes / 28 writes?
- drop `TestRliLookupMetricsOnDataSourceMor*`: no instrumented code branches
on table type (22 writes)
- fold the one distinct SQL case (non-prepped UPDATE exact counts) into the
existing `TestGlobalRecordLevelIndexWithSQL`
- replace `TestRliLookupMetricsReporting` (8 writes, 4 SparkContexts) with
one `TestHoodieMetrics` unit test calling `publishAndRelease`
- drop `TestRliLookupMetricsMultiTable`: its tables have different
`TBL_NAME`s, which separate the keys on their own, so it cannot exercise the
digest
(`TestRecordIndexMetricNames.countersAreScopedByBasePathNotOnlyByTableName`
does)
- drop the AcrossFailedCommit Partitioned twin and the streamer partitioned
arm (covered by `TestRliLookupMetricsOnDataSourcePartitioned`)
- unit side: merge `TestExecutorMetricsGenericity` into
`TestRecordIndexMetricNames`; fold `TestRegistryExecutorLookup` into the
existing `TestRegistry` (3 of 7 pass on master unchanged); merge the two
self-managed-SparkContext `DistributedRegistry` files, dropping the 16-thread
test (`ConcurrentHashMap.compute` serialises per key, so it cannot fail) and
`testRepeatedEvaluationDoubleCounts` (pins Spark, not Hudi)
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsOnSparkSql.scala:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.functional
+
+import org.apache.hudi.DataSourceWriteOptions._
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.config.HoodieIndexConfig
+import org.apache.hudi.metrics.RecordIndexMetricNames
+
+import org.apache.spark.sql.SaveMode
+import org.junit.jupiter.api.{Tag, Test}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+
+/** Record level index lookup counters on the Spark SQL write path. */
+@Tag("functional")
+class TestRliLookupMetricsOnSparkSql extends RliLookupMetricsTestBase {
+
+ private val sqlTable = "rli_lookup_metrics_tbl"
+ private val numSeedRecords = 60
+
+ /**
+ * Seeds a table through the DataSource so the record index exists, then
exposes it to SQL and applies
+ * the index settings as session configs -- index type is a write config,
not a table property.
+ */
+ private def seedTableAndRegisterForSql(): Unit = {
+ doWriteAndValidateDataAndRecordIndex(rliOpts, INSERT_OPERATION_OPT_VAL,
SaveMode.Overwrite,
+ validate = false, numInserts = numSeedRecords)
+
+ spark.sql(s"drop table if exists $sqlTable")
+ spark.sql(s"create table $sqlTable using hudi location '$basePath'")
+
+ spark.sql("set hoodie.write.lock.provider =
org.apache.hudi.client.transaction.lock.InProcessLockProvider")
+ spark.sql(s"set ${HoodieMetadataConfig.ENABLE.key} = true")
+ spark.sql(s"set
${HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key} =
${!isPartitionedRli}")
+ spark.sql(s"set ${HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key}
= $isPartitionedRli")
+ spark.sql(s"set ${HoodieIndexConfig.INDEX_TYPE.key} = " +
+ (if (isPartitionedRli) "RECORD_LEVEL_INDEX" else
"GLOBAL_RECORD_LEVEL_INDEX"))
+
+ clearRliRegistry()
+ }
+
+ /**
+ * The default path. Optimized writes make UPDATE a prepped write, so no
index lookup happens and no
+ * counters are produced. Documented behaviour, asserted so it cannot change
unnoticed.
+ */
+ @Test
+ def testUpdateWithOptimizedWritesPerformsNoLookup(): Unit = {
+ seedTableAndRegisterForSql()
+ spark.sql(s"set ${SPARK_SQL_OPTIMIZED_WRITES.key} = true")
+
+ spark.sql(s"update $sqlTable set rider = 'rider-optimized'")
+
+ val counters = rliCountersFromLatestCommit()
+ report(s"Spark SQL UPDATE, optimized writes ON ($indexLabel) -- expected
empty", counters)
+ assertTrue(counters.isEmpty,
Review Comment:
This assertion has no positive control in the method: the seed is an INSERT
(no lookup), so `counters.isEmpty` also holds if the feature is broken or
mis-gated. Could the test first assert that a preceding upsert did publish (as
`TestRliLookupMetricsOnDataSource:85` does) before asserting the prepped UPDATE
is silent?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/RliLookupMetricsTestBase.scala:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.functional
+
+import org.apache.hudi.DataSourceWriteOptions
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.metrics.Registry
+import org.apache.hudi.config.HoodieIndexConfig
+import org.apache.hudi.metrics.RecordIndexMetricNames
+
+import scala.collection.JavaConverters._
+
+/**
+ * Shared plumbing for the record level index lookup metric tests: index
selection, and reading the counters back the way an operator would -- off the la
Review Comment:
nit: the sentence is cut off.
```suggestion
* Shared plumbing for the record level index lookup metric tests: index
selection, and reading the counters back the way an operator would -- off the
latest completed commit.
```
##########
hudi-common/src/test/java/org/apache/hudi/common/TestRegistryExecutorLookup.java:
##########
@@ -0,0 +1,167 @@
+/*
+ * 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.common;
+
+import org.apache.hudi.common.metrics.ExecutorMetricsContext;
+import org.apache.hudi.common.metrics.LocalRegistry;
+import org.apache.hudi.common.metrics.Registry;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Characterizes how an executor reaches a driver-published registry by name.
*/
+public class TestRegistryExecutorLookup {
Review Comment:
Same as the genericity test: `publishAsDriverDoes` and the bare-name lookups
leave non-zero entries in the process-wide `Registry.REGISTRY_MAP` with no
`@AfterEach`; `TestRegistry.testGetAllMetrics:64` in this package only survives
because its own `flush=true` scrape wipes them. Could this file get the same
cleanup (or be folded into `TestRegistry`)?
##########
hudi-common/src/main/java/org/apache/hudi/common/config/metrics/HoodieMetricsConfig.java:
##########
@@ -109,6 +109,17 @@ public class HoodieMetricsConfig extends HoodieConfig {
.sinceVersion("0.13.0")
.withDocumentation("Enable metrics for locking infra. Useful when
operating in multiwriter mode");
+ public static final ConfigProperty<Boolean> RLI_LOOKUP_METRICS_ENABLE =
ConfigProperty
+ .key(METRIC_PREFIX + ".rli.lookup.enable")
+ .defaultValue(true)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("Collect counters for the record level index lookup
phase (records looked up, "
Review Comment:
The doc reads as engine-neutral, but only the Spark RLI lookup emits: Flink
maps `RECORD_LEVEL_INDEX` to `FlinkInMemoryStateIndex` and the Java client has
no RLI case, so on those engines the drain finds no registry. Two write clients
on the same table in one JVM (OCC, in-process async services) also share one
registry and both snapshot before either releases. Could the doc say "Spark
only" and note that counters are per table per JVM rather than per writer?
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistry.java:
##########
@@ -184,6 +187,131 @@ public void testAddMetricsParallel() {
Assertions.assertEquals(finalExpectedSum, metricCounts.get(METRIC_1));
}
+ @Test
+ public void testSetThrowsOnExecutor() {
+ // Given: a registry registered to the spark context
+ String registryName = REGISTRY_NAME + "_testSetOnExecutor";
+ Registry registry = engineContext.getMetricRegistry("", registryName);
+
+ List<Integer> data = new ArrayList<>();
+ data.add(1);
+
+ // When/Then: set() invoked on an executor must fail - it is
non-commutative under accumulator merges.
+ // The UnsupportedOperationException thrown on the executor surfaces
wrapped in a SparkException.
+ assertFailsOnExecutorWith("DistributedRegistry.set() must not be called
from a Spark executor", () ->
+ engineContext.map(data, value -> {
+ registry.set(METRIC_1, value);
+ return null;
+ }, 1));
+ }
+
+ @Test
+ public void testReleaseThrowsOnExecutor() {
+ // Given: a registry registered to the spark context
+ String registryName = REGISTRY_NAME + "_testReleaseOnExecutor";
+ Registry registry = engineContext.getMetricRegistry("", registryName);
+
+ List<Integer> data = new ArrayList<>();
+ data.add(1);
+
+ // When/Then: release() invoked on an executor must fail - clamping and
eviction are order-dependent
+ // under accumulator merges. The UnsupportedOperationException surfaces
wrapped in a SparkException.
+ assertFailsOnExecutorWith("DistributedRegistry.release() must not be
called from a Spark executor", () ->
+ engineContext.map(data, value -> {
+ registry.release(Collections.singletonMap(METRIC_1, (long) value));
+ return null;
+ }, 1));
+ }
+
+ /**
+ * Asserts the job failed because the executor-side guard fired, not for
some unrelated reason such as a serialization error.
+ */
+ private static void assertFailsOnExecutorWith(String expectedMessage,
Executable executable) {
Review Comment:
nit, feel free to ignore: matching the exception message substring means a
reword of the guard text breaks these two tests without any behaviour change.
Could `assertFailsOnExecutorWith` match `UnsupportedOperationException` on the
cause chain instead?
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metrics/TestExecutorMetricsGenericity.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.metrics;
+
+import org.apache.hudi.common.metrics.ExecutorMetricsContext;
+import org.apache.hudi.common.metrics.LocalRegistry;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The claim this feature exists to support is that adding a class of executor
metric costs a declaration
+ * plus two lines at the emission site. These tests hold that claim to a
measurement rather than an
+ * argument, by collecting a group the shipping code has never heard of.
+ */
+public class TestExecutorMetricsGenericity {
+
+ private static final String BASE_PATH = "file:///tmp/test_generic_metrics";
+
+ /** A class of metric added by a hypothetical future contributor: a name, a
prefix, and no config. */
+ private static final ExecutorMetricGroup STORAGE_CALLS = new
ExecutorMetricGroup() {
+ @Override
+ public String registryName() {
+ return "HoodieStorageCalls";
+ }
+
+ @Override
+ public String commitMetadataPrefix() {
+ return "hoodie.storage.calls.";
+ }
+
+ @Override
+ public String metricAction() {
+ return "storage";
+ }
+
+ @Override
+ public String metricQualifier() {
+ return "calls";
+ }
+
+ @Override
+ public boolean isEnabled(HoodieWriteConfig config) {
+ return true;
+ }
+
+ @Override
+ public String scopedName(String basePath) {
+ return registryName() + ".test";
+ }
+ };
+
+ private static HoodieWriteConfig config() {
+ return
HoodieWriteConfig.newBuilder().withPath(BASE_PATH).forTable("generic_metrics_table").build();
+ }
+
+ private static Registry seed(HoodieWriteConfig cfg) {
+ Registry registry = new
LocalRegistry(STORAGE_CALLS.scopedName(cfg.getBasePath()));
+ Registry.REGISTRY_MAP.put(
Review Comment:
These `put`s go into the process-wide `Registry.REGISTRY_MAP` with no
`@AfterEach`, and `twoGroupsInOneCommitDoNotMix` never releases, so `open=3` /
`tag.lookup_record_index_key_count=11` outlive the class under
`reuseForks=true`; the sibling `TestRecordIndexMetricNames:52` cleans up. Also
the `allMatch` at L117 cannot fail (the map was just filled from a single-group
list). Could this get the same `@AfterEach removeIf`, and the `allMatch` be
dropped?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/RliLookupMetricsTestBase.scala:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.functional
+
+import org.apache.hudi.DataSourceWriteOptions
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.metrics.Registry
+import org.apache.hudi.config.HoodieIndexConfig
+import org.apache.hudi.metrics.RecordIndexMetricNames
+
+import scala.collection.JavaConverters._
+
+/**
+ * Shared plumbing for the record level index lookup metric tests: index
selection, and reading the counters back the way an operator would -- off the la
+ */
+abstract class RliLookupMetricsTestBase extends RecordLevelIndexTestBase {
+
+ /** Overridden by the partitioned subclasses; both variants are separate
closures on separate paths. */
+ protected def isPartitionedRli: Boolean = false
+
+ /**
+ * Table type under test. Tagging is an index-level concern and does not
branch on table type, so MOR
+ * is expected to behave identically -- the MOR subclasses exist to prove
that rather than assume it.
+ */
+ protected def tableTypeOpt: String =
DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL
+
+ protected def indexLabel: String = {
+ val idx = if (isPartitionedRli) "partitioned RLI" else "global RLI"
+ val tt = if (tableTypeOpt ==
DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL) "MOR" else "COW"
+ s"$idx, $tt"
+ }
+
+ /**
+ * `commonOpts` turns the global record index on, so the metadata-partition
flags and the index type
+ * have to be flipped together to select the partitioned variant.
+ */
+ protected def rliOpts: Map[String, String] = {
+ val withTableType = Map(DataSourceWriteOptions.TABLE_TYPE.key ->
tableTypeOpt)
+ if (isPartitionedRli) {
+ commonOpts ++ withTableType ++ Map(
+ HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key ->
"false",
+ HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key -> "true",
+ HoodieIndexConfig.INDEX_TYPE.key -> "RECORD_LEVEL_INDEX")
+ } else {
+ commonOpts ++ withTableType ++ Map(
+ HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key ->
"true",
+ HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key -> "false",
+ HoodieIndexConfig.INDEX_TYPE.key -> "GLOBAL_RECORD_LEVEL_INDEX")
+ }
+ }
+
+ protected def counterKey(caller: String, metric: String): String =
+ RecordIndexMetricNames.COMMIT_METADATA_PREFIX +
RecordIndexMetricNames.key(caller, metric)
+
+ protected def tagKey(metric: String): String =
+ counterKey(RecordIndexMetricNames.CALLER_TAG_LOCATION, metric)
+
+ /**
+ * A caller that looked something up stamps its full counter set, zeros
included, so an absent key means
+ * that caller contributed nothing at all. The default is defensive against
exactly that case.
+ */
+ protected def counterOrZero(counters: Map[String, String], caller: String,
metric: String): Long =
+ counters.getOrElse(counterKey(caller, metric), "0").toLong
+
+ /** The counters as an operator would read them: off the latest completed
commit. */
+ protected def rliCountersFromLatestCommit(): Map[String, String] = {
+ metaClient.reloadActiveTimeline()
+ val lastInstant =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants().lastInstant().get()
+
metaClient.getActiveTimeline.readCommitMetadata(lastInstant).getExtraMetadata.asScala.toMap
+ .filter { case (k, _) =>
k.startsWith(RecordIndexMetricNames.COMMIT_METADATA_PREFIX) }
+ }
+
+ /** Leftover counters from a previous write would otherwise be folded into
the next commit. */
+ protected def clearRliRegistry(): Unit = {
Review Comment:
`clearRliRegistry()` is a no-op at all five call sites: `@TempDir` is per
method, so the base-path digest (and the registry key) is unique per test, and
the seeding write is an INSERT that this suite itself asserts produces no
counters. Could the helper and its calls be deleted, so the tests do not imply
a cross-write leak that the per-method temp dir already rules out?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/RecordIndexLookupMetrics.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.index;
+
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.metrics.DistributedRegistry;
+import org.apache.hudi.metrics.ExecutorMetricGroup;
+import org.apache.hudi.metrics.ExecutorMetricRegistry;
+import org.apache.hudi.metrics.RecordIndexMetricNames;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/** Executor-side emission for the record index lookup counters. */
+public class RecordIndexLookupMetrics {
+
+ /** Set by the read client around its own tagging call, so dedupe traffic is
attributable separately. */
+ private static final ThreadLocal<String> CALLER =
+ ThreadLocal.withInitial(() ->
RecordIndexMetricNames.CALLER_TAG_LOCATION);
+
+ private RecordIndexLookupMetrics() {
+ }
+
+ public static String currentCaller() {
+ return CALLER.get();
+ }
+
+ /** Restore rather than clear, so a nested tagging call does not reset the
label. */
+ public static String setCaller(String caller) {
+ String previous = CALLER.get();
+ CALLER.set(caller);
+ return previous;
+ }
+
+ public static void restoreCaller(String previous) {
+ CALLER.set(previous);
+ }
+
+ /**
+ * The registries a lookup task collects into, keyed by bare name. Includes
every entry on
+ * {@link ExecutorMetricRegistry}. Delivery is by closure capture, which is
deterministic; resolution is
+ * by name, which lets code below the write API take part without a
signature change.
+ */
+ public static Map<String, Registry> resolveBundle(HoodieEngineContext
context, HoodieWriteConfig config) {
+ return resolveBundle(context, config,
Arrays.asList(ExecutorMetricRegistry.values()));
+ }
+
+ /** Visible for testing the bundle against a group the enum does not ship
with. */
+ public static Map<String, Registry> resolveBundle(HoodieEngineContext
context, HoodieWriteConfig config,
+ Collection<? extends
ExecutorMetricGroup> groups) {
+ Map<String, Registry> bundle = new HashMap<>();
+ for (ExecutorMetricGroup metricRegistry : groups) {
+ if (!metricRegistry.isEnabled(config)) {
+ continue;
+ }
+ Registry registry = context.getMetricRegistry(config.getTableName(),
+ metricRegistry.scopedName(config.getBasePath()));
+ // Only the accumulator-backed registry aggregates back to the driver,
so anything else is left out
+ // rather than bound: a bound LocalRegistry would collect on the
executor and be dropped on the floor,
+ // whereas leaving it out makes the lookup resolve to a no-op that
reports nothing.
+ if (registry instanceof DistributedRegistry) {
+ bundle.put(metricRegistry.registryName(), registry);
+ }
+ }
+ return bundle.isEmpty() ? Collections.emptyMap() : bundle;
+ }
+
+ /**
+ * Records one shard's lookup outcome. Counts records rather than distinct
keys, so
+ * {@code hits + misses == records_looked_up} holds when a batch repeats a
key. Membership is tested
+ * against the found set, bounded by the hit count, not the asked-about set,
bounded by shard size.
+ *
+ * @param keysLookedUp every record key routed to this shard
+ * @param foundKeys the subset present in the index
+ * @param elapsedMs wall-clock spent reading this shard
+ */
+ public static void recordShardLookup(String caller, Collection<String>
keysLookedUp,
Review Comment:
The query read path (`PartitionedRecordLevelIndexSupport.scala:107`, 1-arg
ctor, `caller = null`) and the disabled path both still start the timer, run
the O(n) `found::contains` scan and build `"null.<metric>"` keys, all into
`NoOpRegistry`. Could this return before the scan when the registry is a no-op
(or, if the bundle is passed in directly, when it is null)?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/ExecutorMetricRegistry.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.metrics;
+
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.function.Predicate;
+
+/**
+ * Every class of executor-collected metric, and the only thing a new one is
added to. The driver must
+ * declare it up front because an {@code AccumulatorV2} must be registered
with the {@code SparkContext}
+ * before a task can contribute; the bundle sent to executors and the commit
drain both iterate this.
+ */
+public enum ExecutorMetricRegistry implements ExecutorMetricGroup {
+
+ RECORD_INDEX_LOOKUP(
+ "HoodieRecordIndexLookup",
+ "hoodie.rli.lookup.",
+ "rli",
+ "lookup",
+ HoodieWriteConfig::isRecordIndexLookupMetricsEnabled);
+
+ private final String registryName;
+ private final String commitMetadataPrefix;
+ private final String metricAction;
+ private final String metricQualifier;
+ private final Predicate<HoodieWriteConfig> enabled;
+
+ ExecutorMetricRegistry(String registryName, String commitMetadataPrefix,
String metricAction,
+ String metricQualifier, Predicate<HoodieWriteConfig>
enabled) {
+ this.registryName = registryName;
+ this.commitMetadataPrefix = commitMetadataPrefix;
+ this.metricAction = metricAction;
+ this.metricQualifier = metricQualifier;
+ this.enabled = enabled;
+ }
+
+ /** The bare name emitting code passes to {@link
Registry#getRegistry(String)}. */
+ @Override
+ public String registryName() {
+ return registryName;
+ }
+
+ @Override
+ public String commitMetadataPrefix() {
+ return commitMetadataPrefix;
+ }
+
+ @Override
+ public String metricAction() {
+ return metricAction;
+ }
+
+ @Override
+ public String metricQualifier() {
+ return metricQualifier;
+ }
+
+ /** Gating here, rather than in the drain, is what lets a new class of
metric need no new config. */
+ @Override
+ public boolean isEnabled(HoodieWriteConfig config) {
+ return enabled.test(config);
+ }
+
+ /**
+ * Driver-side {@code REGISTRY_MAP} key. Table name alone is not an
identity: two tables can share one
+ * and would then share a registry. Executors use the bare {@link
#registryName()}.
+ */
+ @Override
+ public String scopedName(String basePath) {
Review Comment:
Because this name contains a `.`, `Registry.getAllMetrics(true, true)` (run
by `Metrics.shutdown`, i.e. after every DataSource write with metrics on) skips
the common prefix and republishes any leftover counters as
`<table>.HoodieRecordIndexLookup.<12-hex>.tag.*`, a per-base-path metric name,
while also clearing the registry. Could this registry be excluded from the
common scrape, or given a digest-free name for it?
##########
hudi-io/src/main/java/org/apache/hudi/common/metrics/Registry.java:
##########
@@ -190,6 +199,28 @@ static void setRegistries(Collection<Registry> registries)
{
*/
void set(String name, long value);
+ /**
+ * Subtract a set of counts previously read out of this registry, clamping
every counter at zero.
+ *
+ * Used to hand a batch of counters over to a consumer that owns them from
then on -- the commit-boundary
+ * drain for the record index lookup counters -- without discarding whatever
arrived after they were read.
+ *
+ * Clamping is what distinguishes this from {@code add(name, -value)}. The
registry can be emptied
+ * underneath a caller by an unrelated destructive scrape ({@link
#getAllMetrics(boolean, boolean)} with
+ * {@code flush=true} clears every registry in the process), and an
unbounded subtraction would then leave
+ * negative counters behind for good.
+ *
+ * The default is a best-effort read-modify-write. Implementations able to
do this atomically should
+ * override it, and should drop counters that reach zero rather than leaving
them at zero, so a registry
+ * nobody is writing to reads as empty.
+ *
+ * @param counts the counts to release, as returned by {@link
#getAllCounts(boolean)}.
+ */
+ default void release(Map<String, Long> counts) {
Review Comment:
This default and the `LocalRegistry` override are unreachable in production:
the only `release` caller is `ExecutorMetrics:94`, whose registry is always a
`DistributedRegistry` (Flink/Java never create the drain key), and only
`TestDistributedRegistry` exercises `release`. Could the default be dropped
(leaving `release` on `DistributedRegistry`, or abstract), or a `LocalRegistry`
test added if it is meant to be supported?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/HoodieSparkEngineContext.java:
##########
@@ -278,11 +279,43 @@ public String getApplicationId() {
return javaSparkContext.sc().applicationId();
}
+ /**
+ * Drops a registry from both process-wide maps. Only for tests that create
their own SparkContexts:
+ * without it they leave accumulators bound to stopped contexts behind for
whatever runs next in the
+ * same JVM.
+ */
+ @VisibleForTesting
+ public static void removeMetricRegistryForTesting(String tableName, String
registryName) {
Review Comment:
Two things here. (1) In local mode `setRegistries` also indexes the same
registry under `Registry.makeKey("", prefixedName)`, so this remover and the
staleness branch below evict only the `table::registry` key and the accumulator
bound to the stopped context survives under `::table.registry`. (2) This is the
only `*ForTesting` public static in `src/main`; the convention is
`@VisibleForTesting` on a normally named method. Could it also drop the
`""`-keyed entry and be renamed `removeMetricRegistry`?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/ExecutorMetricGroup.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.metrics;
+
+import org.apache.hudi.config.HoodieWriteConfig;
+
+/**
+ * One class of executor-collected metric. Implemented by {@link
ExecutorMetricRegistry}, which enumerates
+ * the ones that ship; kept an interface so the collection machinery can be
exercised against a group it
+ * does not know about.
+ */
+public interface ExecutorMetricGroup {
Review Comment:
This interface has one production implementer (the enum, one constant), both
`groups` overloads are only reachable from tests, and the 3-arg
`RecordIndexLookupMetrics.resolveBundle` (L72) has no caller at all. #19063
notes the next executor-metric users (index and MDT read paths) need histogram
primitives first, so the second group is likely a different shape. Could we
inline this to one concrete class now (also dropping
`TestExecutorMetricsGenericity` and the test-only public helpers
`RecordIndexMetricNames.registryName` / `DrainedCounters.isEmpty`), and extract
the abstraction when a second group lands?
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistry.java:
##########
@@ -184,6 +187,131 @@ public void testAddMetricsParallel() {
Assertions.assertEquals(finalExpectedSum, metricCounts.get(METRIC_1));
}
+ @Test
+ public void testSetThrowsOnExecutor() {
+ // Given: a registry registered to the spark context
+ String registryName = REGISTRY_NAME + "_testSetOnExecutor";
+ Registry registry = engineContext.getMetricRegistry("", registryName);
+
+ List<Integer> data = new ArrayList<>();
+ data.add(1);
+
+ // When/Then: set() invoked on an executor must fail - it is
non-commutative under accumulator merges.
+ // The UnsupportedOperationException thrown on the executor surfaces
wrapped in a SparkException.
+ assertFailsOnExecutorWith("DistributedRegistry.set() must not be called
from a Spark executor", () ->
+ engineContext.map(data, value -> {
+ registry.set(METRIC_1, value);
+ return null;
+ }, 1));
+ }
+
+ @Test
+ public void testReleaseThrowsOnExecutor() {
+ // Given: a registry registered to the spark context
+ String registryName = REGISTRY_NAME + "_testReleaseOnExecutor";
+ Registry registry = engineContext.getMetricRegistry("", registryName);
+
+ List<Integer> data = new ArrayList<>();
+ data.add(1);
+
+ // When/Then: release() invoked on an executor must fail - clamping and
eviction are order-dependent
+ // under accumulator merges. The UnsupportedOperationException surfaces
wrapped in a SparkException.
+ assertFailsOnExecutorWith("DistributedRegistry.release() must not be
called from a Spark executor", () ->
+ engineContext.map(data, value -> {
+ registry.release(Collections.singletonMap(METRIC_1, (long) value));
+ return null;
+ }, 1));
+ }
+
+ /**
+ * Asserts the job failed because the executor-side guard fired, not for
some unrelated reason such as a serialization error.
+ */
+ private static void assertFailsOnExecutorWith(String expectedMessage,
Executable executable) {
+ SparkException thrown = Assertions.assertThrows(SparkException.class,
executable);
+ StringBuilder chain = new StringBuilder();
+ for (Throwable t = thrown; t != null; t = t.getCause()) {
+ chain.append(t).append('\n');
+ if (t.getCause() == t) {
+ break;
+ }
+ }
+ Assertions.assertTrue(chain.toString().contains(expectedMessage),
+ "expected the executor-side guard to fail the job, got: " + chain);
+ }
+
+ @Test
+ public void testSetOnDriverSucceeds() {
+ // set() on the driver (no TaskContext) remains supported.
+ DistributedRegistry registry = new DistributedRegistry(REGISTRY_NAME +
"_testSetOnDriver");
+ registry.set(METRIC_1, 42);
+ Assertions.assertEquals(42, registry.getAllCounts().get(METRIC_1));
+ }
+
+ @Test
+ public void testGetMetricRegistryReplacesNonDistributedRegistry() {
Review Comment:
This test never enters the branch it names: `getMetricRegistry` removes the
seeded `LocalRegistry` at `HoodieSparkEngineContext:308` before
`getRegistryOfClass` runs, so `computeIfAbsent` creates a fresh
`DistributedRegistry` and the `!(instanceof)` fallback at L310 stays uncovered.
Could the test seed via `Registry.getRegistryOfClass` from a second thread
released after the remove (or the branch be dropped if the instance-scoped
registry lands)?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/RecordIndexLookupMetrics.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.index;
+
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.metrics.DistributedRegistry;
+import org.apache.hudi.metrics.ExecutorMetricGroup;
+import org.apache.hudi.metrics.ExecutorMetricRegistry;
+import org.apache.hudi.metrics.RecordIndexMetricNames;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/** Executor-side emission for the record index lookup counters. */
+public class RecordIndexLookupMetrics {
+
+ /** Set by the read client around its own tagging call, so dedupe traffic is
attributable separately. */
+ private static final ThreadLocal<String> CALLER =
+ ThreadLocal.withInitial(() ->
RecordIndexMetricNames.CALLER_TAG_LOCATION);
+
+ private RecordIndexLookupMetrics() {
+ }
+
+ public static String currentCaller() {
+ return CALLER.get();
+ }
+
+ /** Restore rather than clear, so a nested tagging call does not reset the
label. */
+ public static String setCaller(String caller) {
+ String previous = CALLER.get();
+ CALLER.set(caller);
+ return previous;
+ }
+
+ public static void restoreCaller(String previous) {
+ CALLER.set(previous);
+ }
+
+ /**
+ * The registries a lookup task collects into, keyed by bare name. Includes
every entry on
+ * {@link ExecutorMetricRegistry}. Delivery is by closure capture, which is
deterministic; resolution is
+ * by name, which lets code below the write API take part without a
signature change.
+ */
+ public static Map<String, Registry> resolveBundle(HoodieEngineContext
context, HoodieWriteConfig config) {
+ return resolveBundle(context, config,
Arrays.asList(ExecutorMetricRegistry.values()));
+ }
+
+ /** Visible for testing the bundle against a group the enum does not ship
with. */
+ public static Map<String, Registry> resolveBundle(HoodieEngineContext
context, HoodieWriteConfig config,
+ Collection<? extends
ExecutorMetricGroup> groups) {
+ Map<String, Registry> bundle = new HashMap<>();
+ for (ExecutorMetricGroup metricRegistry : groups) {
+ if (!metricRegistry.isEnabled(config)) {
+ continue;
+ }
+ Registry registry = context.getMetricRegistry(config.getTableName(),
+ metricRegistry.scopedName(config.getBasePath()));
+ // Only the accumulator-backed registry aggregates back to the driver,
so anything else is left out
+ // rather than bound: a bound LocalRegistry would collect on the
executor and be dropped on the floor,
+ // whereas leaving it out makes the lookup resolve to a no-op that
reports nothing.
+ if (registry instanceof DistributedRegistry) {
+ bundle.put(metricRegistry.registryName(), registry);
+ }
+ }
+ return bundle.isEmpty() ? Collections.emptyMap() : bundle;
+ }
+
+ /**
+ * Records one shard's lookup outcome. Counts records rather than distinct
keys, so
+ * {@code hits + misses == records_looked_up} holds when a batch repeats a
key. Membership is tested
+ * against the found set, bounded by the hit count, not the asked-about set,
bounded by shard size.
+ *
+ * @param keysLookedUp every record key routed to this shard
+ * @param foundKeys the subset present in the index
+ * @param elapsedMs wall-clock spent reading this shard
+ */
+ public static void recordShardLookup(String caller, Collection<String>
keysLookedUp,
+ Collection<String> foundKeys, long
elapsedMs) {
+ if (keysLookedUp.isEmpty()) {
+ return;
+ }
+ Registry registry =
Registry.getRegistry(RecordIndexMetricNames.REGISTRY_NAME);
Review Comment:
Both callers of `recordShardLookup` already hold `metricsBundle` as a field
(`SparkMetadataTableGlobalRecordLevelIndex:214`,
`PartitionedRecordIndexFileGroupLookupFunction:92`), so resolving by name here
can only return `metricsBundle.get(REGISTRY_NAME)`; nothing shipped emits
without the bundle in hand. Passing that `Registry` in directly (null ->
return) would let `ExecutorMetricsContext`, `NoOpRegistry`,
`TestRegistryExecutorLookup`, the `HoodieSparkTable` binding and the
`Registry.getRegistry` contract change go (about 330 lines, both hudi-io
additions) with identical behaviour. Could we take that shape and defer name
resolution until an emitter that lacks the bundle exists?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/PartitionedRecordIndexFileGroupLookupFunction.java:
##########
@@ -44,36 +47,57 @@ public class PartitionedRecordIndexFileGroupLookupFunction
implements PairFlatMapFunction<Iterator<Pair<String, String>>, String,
HoodieRecordGlobalLocation> {
private final HoodieTableMetadata metadataTable;
+ // Empty when no counters should be collected; see
RecordIndexLookupMetrics#resolveBundle.
+ private final Map<String, Registry> metricsBundle;
+ private final String caller;
+ /** Uninstrumented, for the query-side read path. */
public PartitionedRecordIndexFileGroupLookupFunction(HoodieTableMetadata
metadataTable) {
+ this(metadataTable, Collections.emptyMap(), null);
+ }
+
+ public PartitionedRecordIndexFileGroupLookupFunction(HoodieTableMetadata
metadataTable,
+ Map<String, Registry>
metricsBundle, String caller) {
this.metadataTable = metadataTable;
+ this.metricsBundle = metricsBundle;
+ this.caller = caller;
}
@Override
public Iterator<Tuple2<String, HoodieRecordGlobalLocation>>
call(Iterator<Pair<String, String>> partitionPathRecordKeyIterator) {
- String partitionName = null;
- List<String> keysToLookup = new ArrayList<>();
- while (partitionPathRecordKeyIterator.hasNext()) {
- Pair<String, String> partitionPathRecordKey =
partitionPathRecordKeyIterator.next();
- keysToLookup.add(partitionPathRecordKey.getRight());
- if (partitionName == null) {
- partitionName = partitionPathRecordKey.getLeft();
+ // Bound for the whole task so a metric raised deeper in the lookup
resolves here too.
Review Comment:
nit, feel free to ignore: the binding is released in the `finally` before
Spark consumes the returned iterator, so it covers `call()` (the shard read),
not the whole task.
```suggestion
// Bound for the duration of the shard read so a metric raised deeper in
the lookup resolves here too.
```
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestRliLookupMetricsReporting.scala:
##########
@@ -0,0 +1,114 @@
+/*
+ * 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.functional
+
+import org.apache.hudi.DataSourceWriteOptions._
+import org.apache.hudi.common.config.metrics.HoodieMetricsConfig
+import org.apache.hudi.metrics.{ExecutorMetricRegistry, RecordIndexMetricNames}
+
+import org.apache.spark.sql.SaveMode
+import org.junit.jupiter.api.{Tag, Test}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+
+import java.io.{ByteArrayOutputStream, PrintStream}
+
+/** The counters must reach a live metrics reporter, not only commit metadata.
*/
+@Tag("functional")
+class TestRliLookupMetricsReporting extends RliLookupMetricsTestBase {
+
+ private def metricsOpts: Map[String, String] = rliOpts ++ Map(
+ HoodieMetricsConfig.TURN_METRICS_ON.key -> "true",
+ HoodieMetricsConfig.METRICS_REPORTER_TYPE_VALUE.key -> "CONSOLE")
+
+ /** Captures stdout for the duration of the write. */
+ private def captureStdout(body: => Unit): String = {
+ val buffer = new ByteArrayOutputStream()
+ val original = System.out
+ try {
+ System.setOut(new PrintStream(buffer, true, "UTF-8"))
+ body
+ } finally {
+ System.setOut(original)
+ }
+ buffer.toString("UTF-8")
+ }
+
+ @Test
+ def testCountersReachBothCommitMetadataAndTheReporter(): Unit = {
+ val numUpdates = 12
+
+ doWriteAndValidateDataAndRecordIndex(metricsOpts,
INSERT_OPERATION_OPT_VAL, SaveMode.Overwrite,
+ validate = false, numInserts = 50)
+ clearRliRegistry()
+
+ val stdout = captureStdout {
+ doWriteAndValidateDataAndRecordIndex(metricsOpts,
UPSERT_OPERATION_OPT_VAL, SaveMode.Append,
+ validate = false, numUpdates = numUpdates)
+ }
+
+ // Sink 1 -- the timeline.
+ val counters = rliCountersFromLatestCommit()
+ report(s"Reporter test ($indexLabel) -- commit metadata", counters)
+ assertTrue(counters.nonEmpty, "the commit must still carry the counters
when a reporter is configured")
+ assertEquals(numUpdates.toString,
counters(tagKey(RecordIndexMetricNames.KEY_HIT_COUNT)))
+ assertEquals((numUpdates + 1).toLong, assertSumInvariant(counters,
RecordIndexMetricNames.CALLER_TAG_LOCATION))
+
+ // Sink 2 -- the reporter. Gauge names are
<prefix>.rli.lookup.<caller>.<metric>.
+ val gaugePrefix =
s"${ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.metricAction}.${ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.metricQualifier}"
+ val reported = stdout.linesIterator.filter(_.contains(gaugePrefix)).toSeq
+
+ println(s"\n===== Reporter test ($indexLabel) -- ConsoleMetricsReporter
output =====")
+ if (reported.isEmpty) println(" (no rli.lookup gauges printed)") else
reported.foreach(l => println(s" ${l.trim}"))
+
println("=======================================================================\n")
+
+ assertTrue(reported.nonEmpty,
+ s"ConsoleMetricsReporter must publish the '$gaugePrefix' gauges; the
drain feeds the reporter and " +
+ "commit metadata from a single read, so finding them in the commit but
not here means the " +
+ "reporter sink regressed")
+ Seq(RecordIndexMetricNames.KEY_HIT_COUNT,
RecordIndexMetricNames.KEY_MISS_COUNT,
+ RecordIndexMetricNames.KEY_COUNT,
RecordIndexMetricNames.SHARDS_READ).foreach { metric =>
+ val name =
s"$gaugePrefix.${RecordIndexMetricNames.key(RecordIndexMetricNames.CALLER_TAG_LOCATION,
metric)}"
+ assertTrue(stdout.contains(name), s"reporter output must contain the
gauge '$name'")
Review Comment:
This checks that the gauge name was printed, never a value, so a regression
that publishes zeros or a stale map to `Metrics.registerGauges` passes. The
console output has `value = N` on the next line and the expected value is
derivable (`numUpdates + 1`). Could the test assert the value too (or, per the
consolidation note, move to a `TestHoodieMetrics` unit test that reads the
gauge directly)?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/RliLookupMetricsTestBase.scala:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.functional
+
+import org.apache.hudi.DataSourceWriteOptions
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.metrics.Registry
+import org.apache.hudi.config.HoodieIndexConfig
+import org.apache.hudi.metrics.RecordIndexMetricNames
+
+import scala.collection.JavaConverters._
+
+/**
+ * Shared plumbing for the record level index lookup metric tests: index
selection, and reading the counters back the way an operator would -- off the la
+ */
+abstract class RliLookupMetricsTestBase extends RecordLevelIndexTestBase {
+
+ /** Overridden by the partitioned subclasses; both variants are separate
closures on separate paths. */
+ protected def isPartitionedRli: Boolean = false
+
+ /**
+ * Table type under test. Tagging is an index-level concern and does not
branch on table type, so MOR
+ * is expected to behave identically -- the MOR subclasses exist to prove
that rather than assume it.
+ */
+ protected def tableTypeOpt: String =
DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL
+
+ protected def indexLabel: String = {
+ val idx = if (isPartitionedRli) "partitioned RLI" else "global RLI"
+ val tt = if (tableTypeOpt ==
DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL) "MOR" else "COW"
+ s"$idx, $tt"
+ }
+
+ /**
+ * `commonOpts` turns the global record index on, so the metadata-partition
flags and the index type
+ * have to be flipped together to select the partitioned variant.
+ */
+ protected def rliOpts: Map[String, String] = {
+ val withTableType = Map(DataSourceWriteOptions.TABLE_TYPE.key ->
tableTypeOpt)
+ if (isPartitionedRli) {
+ commonOpts ++ withTableType ++ Map(
+ HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key ->
"false",
+ HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key -> "true",
+ HoodieIndexConfig.INDEX_TYPE.key -> "RECORD_LEVEL_INDEX")
+ } else {
+ commonOpts ++ withTableType ++ Map(
+ HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key ->
"true",
+ HoodieMetadataConfig.RECORD_LEVEL_INDEX_ENABLE_PROP.key -> "false",
+ HoodieIndexConfig.INDEX_TYPE.key -> "GLOBAL_RECORD_LEVEL_INDEX")
+ }
+ }
+
+ protected def counterKey(caller: String, metric: String): String =
+ RecordIndexMetricNames.COMMIT_METADATA_PREFIX +
RecordIndexMetricNames.key(caller, metric)
+
+ protected def tagKey(metric: String): String =
+ counterKey(RecordIndexMetricNames.CALLER_TAG_LOCATION, metric)
+
+ /**
+ * A caller that looked something up stamps its full counter set, zeros
included, so an absent key means
+ * that caller contributed nothing at all. The default is defensive against
exactly that case.
+ */
+ protected def counterOrZero(counters: Map[String, String], caller: String,
metric: String): Long =
+ counters.getOrElse(counterKey(caller, metric), "0").toLong
+
+ /** The counters as an operator would read them: off the latest completed
commit. */
+ protected def rliCountersFromLatestCommit(): Map[String, String] = {
+ metaClient.reloadActiveTimeline()
+ val lastInstant =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants().lastInstant().get()
+
metaClient.getActiveTimeline.readCommitMetadata(lastInstant).getExtraMetadata.asScala.toMap
+ .filter { case (k, _) =>
k.startsWith(RecordIndexMetricNames.COMMIT_METADATA_PREFIX) }
+ }
+
+ /** Leftover counters from a previous write would otherwise be folded into
the next commit. */
+ protected def clearRliRegistry(): Unit = {
+ Registry.REGISTRY_MAP.asScala.foreach {
+ case (key, registry) => if
(key.contains(RecordIndexMetricNames.REGISTRY_NAME)) registry.clear()
+ }
+ }
+
+ /** Asserts the core invariant and returns the looked-up count. */
+ protected def assertSumInvariant(counters: Map[String, String], caller:
String): Long = {
+ val lookedUp = counterOrZero(counters, caller,
RecordIndexMetricNames.KEY_COUNT)
+ val hits = counterOrZero(counters, caller,
RecordIndexMetricNames.KEY_HIT_COUNT)
+ val misses = counterOrZero(counters, caller,
RecordIndexMetricNames.KEY_MISS_COUNT)
+ org.junit.jupiter.api.Assertions.assertEquals(lookedUp, hits + misses,
+ s"hits + misses must account for every key looked up by '$caller'")
+ // A caller that looked something up must also report the time it took, or
the timing metric is
+ // silently absent on paths nobody checked. Zero is allowed: a shard read
can round below a millisecond.
+ if (lookedUp > 0) {
+ org.junit.jupiter.api.Assertions.assertTrue(
+ counters.contains(counterKey(caller,
RecordIndexMetricNames.LOOKUP_TIME)),
+ s"'$caller' looked up $lookedUp keys but reported no
${RecordIndexMetricNames.LOOKUP_TIME}; " +
+ s"counters were ${counters.keys.toSeq.sorted.mkString(", ")}")
+ org.junit.jupiter.api.Assertions.assertTrue(
+ counterOrZero(counters, caller, RecordIndexMetricNames.LOOKUP_TIME) >=
0L,
Review Comment:
nit, feel free to ignore: `>= 0L` on an unsigned counter cannot fail (the
`contains` above is the real check), and the `report(...)` helpers at L124-134
are `println`s that assert nothing, run 30+ times into CI logs (13 `println`s
exist across the other 72 functional Scala files combined). Could both go,
relying on the assertion messages that already interpolate `got $counters`?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/ExecutorMetrics.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.metrics;
+
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Commit-boundary drain for executor-collected metrics, generic over {@link
ExecutorMetricRegistry}. On
+ * the shared commit path, so it covers Spark DataSource, Spark SQL and
DeltaStreamer alike.
+ */
+public class ExecutorMetrics {
+
+ private ExecutorMetrics() {
+ }
+
+ /**
+ * Snapshots into commit metadata without consuming. Split from {@link
#publishAndRelease} so a commit
+ * that never lands neither loses its counters nor publishes gauges for
rolled-back work. An all-zero
+ * registry is skipped to keep residue off the timeline; zeros are otherwise
kept, since an explicit
+ * {@code misses=0} is meaningful.
+ */
+ public static DrainedCounters snapshotIntoCommitMetadata(Map<String, String>
commitMetadata,
+ HoodieWriteConfig
config) {
+ return snapshotIntoCommitMetadata(commitMetadata, config,
Arrays.asList(ExecutorMetricRegistry.values()));
+ }
+
+ /** Visible for testing the collection machinery against a group it does not
ship with. */
+ static DrainedCounters snapshotIntoCommitMetadata(Map<String, String>
commitMetadata,
+ HoodieWriteConfig config,
+ Collection<? extends
ExecutorMetricGroup> groups) {
+ List<Drained> drained = new ArrayList<>();
+ for (ExecutorMetricGroup metricRegistry : groups) {
+ if (!metricRegistry.isEnabled(config)) {
+ continue;
+ }
+ Registry registry = Registry.REGISTRY_MAP.get(
+ Registry.makeKey(config.getTableName(),
metricRegistry.scopedName(config.getBasePath())));
+ if (registry == null) {
+ continue;
+ }
+ Map<String, Long> counts = new HashMap<>();
+ boolean recordedSomething = false;
+ for (Map.Entry<String, Long> counter :
registry.getAllCounts(false).entrySet()) {
+ if (counter.getValue() == null) {
Review Comment:
nit, feel free to ignore: this branch is unreachable.
`LocalRegistry.getAllCounts` boxes a primitive `long`, and
`DistributedRegistry`'s map is a `ConcurrentHashMap`, which forbids null
values. Could it go?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkTable.java:
##########
@@ -142,9 +146,25 @@ protected Option<HoodieTableMetadataWriter>
getMetadataWriter(
return Option.empty();
}
+ /**
+ * Carries the task thread's state onto the pool threads Hudi hands record
writing to.
+ *
+ * <p>The metrics binding travels alongside the {@link TaskContext} for the
same reason the
+ * TaskContext does: both are thread-local, and {@code
BoundedInMemoryExecutor} and
+ * {@code DisruptorExecutor} run on threads the task did not create. Without
this, a metric raised
+ * while writing records resolves against an unbound thread and is discarded.
+ *
+ * <p>Nothing unbinds afterwards. These pools are created and shut down per
operation, so the thread
+ * does not outlive the binding; a pool that were shared across tables would
need an unbind hook on
+ * the other side, which no current executor provides.
+ */
@Override
public Runnable getPreExecuteRunnable() {
final TaskContext taskContext = TaskContext.get();
- return () -> TaskContext$.MODULE$.setTaskContext(taskContext);
+ final Map<String, Registry> metricsBinding =
ExecutorMetricsContext.capture();
+ return () -> {
+ TaskContext$.MODULE$.setTaskContext(taskContext);
+ ExecutorMetricsContext.bind(metricsBinding);
Review Comment:
Confirmed on both counts. Every `getPreExecuteRunnable` call site is
write-stage (`HoodieMergeHelper:154`, `SparkLazyInsertIterable:77`, the
bootstrap handlers,
`SingleSparkJobConsistentHashingExecutionStrategy:132/217`), a different stage
from the lookup, and both lookup functions `unbind` in a `finally`, so
`capture()` returns `emptyMap` and the only effect of `bind(emptyMap)` is
`isBound() == true`, which turns every `Registry.getRegistry` miss on those
pool threads into `NoOpRegistry`. Inert today (no production `getRegistry` runs
on executors) and untested. Could it be dropped until something actually emits
from a writer-pool thread?
--
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]
