rahil-c commented on code in PR #19575: URL: https://github.com/apache/hudi/pull/19575#discussion_r3859015366
########## 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: Fixed. `TBL_NAME` has no default and `Builder.validate()` only requires `BASE_PATH`, so a config built without `forTable()` reached `getMetricRegistry` with a null name and NPE'd on `tableName.isEmpty()`. Took the suggestion, skipping collection when the name is null or empty. ########## 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: Fixed, and the point generalises. Both guards now `LOG.warn` and return instead of throwing, and the whole drain at the commit boundary is wrapped in a catch-all that logs and moves on. It runs after the commit has landed, so no reporting problem should be able to take a completed write down with it -- same reason `Metrics.registerGauge` already swallows. The two tests that asserted the throw now assert the behaviour that actually matters: the job succeeds, and `set()` records nothing while `release()` leaves the counters alone. That also drops the exception-message matching you flagged separately. ########## 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: Fixed both. The remover now also drops the `Registry.makeKey("", prefixedName)` entry, and is renamed `removeMetricRegistry` with `@VisibleForTesting`. For what it is worth, this reproduced exactly as described: instrumenting the drain and dumping `REGISTRY_MAP` shows both `hoodie_test::HoodieRecordIndexLookup.<digest>` and `::hoodie_test.HoodieRecordIndexLookup.<digest>` present simultaneously. ########## 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: Fixed. `recordShardLookup` now returns as soon as the resolved registry is a `NoOpRegistry`, before the O(keys) hit scan. That covers the query read path and the disabled path. The `null.<metric>` keys are gone too, since caller tagging has been removed from the PR entirely. ########## 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: Fixed -- you are right that it cannot be null (`LocalRegistry` boxes a primitive `long`, and `ConcurrentHashMap` forbids null values). Simplified to `allMatch(value -> value == 0L)`. -- 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]
