rahil-c commented on code in PR #19575: URL: https://github.com/apache/hudi/pull/19575#discussion_r3874972540
########## hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/RecordIndexLookupMetrics.java: ########## @@ -0,0 +1,99 @@ +/* + * 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.ExecutorMetricRegistry; +import org.apache.hudi.metrics.RecordIndexMetricNames; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; + +/** Executor-side emission for the record index lookup counters. */ +public class RecordIndexLookupMetrics { + + private RecordIndexLookupMetrics() { + } + + /** + * The registry a lookup task collects into, or null when nothing should be collected. Captured in the + * lookup closure and passed to {@link #recordShardLookup}, so delivery is by closure capture rather + * than by name. + * + * <p>Requires the reporter to be on as well: with {@code hoodie.metrics.on} off there is nowhere to + * publish, and collecting would register an accumulator and scan every shard for nothing. + */ + public static Registry resolveRegistry(HoodieEngineContext context, HoodieWriteConfig config) { + if (!config.isMetricsOn() || !ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.isEnabled(config)) { + return null; + } + // TBL_NAME has no default and Builder.validate() only requires BASE_PATH, so a config built without + // forTable() reaches here with a null name. getMetricRegistry dereferences it immediately. + String tableName = config.getTableName(); + if (tableName == null || tableName.isEmpty()) { + return null; + } + Registry registry = context.getMetricRegistry(tableName, + ExecutorMetricRegistry.RECORD_INDEX_LOOKUP.scopedName(config.getBasePath())); + // Only the accumulator-backed registry aggregates back to the driver. A LocalRegistry here would + // collect on the executor and be dropped on the floor, so report nothing instead. + if (!(registry instanceof DistributedRegistry)) { + return null; + } + // Runs on the driver before the closure ships, so anything held now predates this write. Publishing + // releases what it reports, which leaves a non-empty registry only after an attempt that never + // committed. Paths that tear metrics down between writes lose it anyway; Spark SQL DML and StreamSync + // do not, and this commit must not report work it did not do. + registry.clear(); Review Comment: One scoping point that I think bounds it. Every piece of state involved is per-JVM: - `Registry.REGISTRY_MAP` is an interface field, so implicitly `public static final` - `HoodieSparkEngineContext.DISTRIBUTED_REGISTRY_MAP` is `private static final` - the counters are an instance field on the `DistributedRegistry`, and merging goes through the task-result envelope of the `SparkContext` it was registered with Nothing is persisted outside the process either. The commit-metadata sink was removed earlier in this PR, so the only outbound path is `registerGauges`. So two write clients in one JVM do share a bucket and can clear and publish each other's counts, exactly as you describe. Two separate applications writing the same table cannot: separate drivers, separate static maps, separate accumulators, and an executor in one application has no route to the other's accumulator. On disabling for multiwriter: the gate would have to key off `WriteConcurrencyMode.supportsMultiWriter()`, which cannot distinguish separate-application multiwriter from in-process. It would switch the feature off for deployments that are not affected, including the one that asked for it, while the case it guards is the narrower in-process one. On scoping: instant-based scoping is not reachable from where the registry is resolved. `lookupRecords(records, context, hoodieTable, fileGroupSize)` carries no instant, and `HoodieTable` takes it as a parameter rather than holding it. A per-writer key is reachable in about eight lines, by stamping a UUID into the write config at client construction and including it in the registry key. The cost is that the key can then only be reconstructed from a live client config, which breaks the test helper that seeds the abandoned-attempt regression case. Proposal: take this as a follow-up scoped on whether in-process concurrent writers are actually in use, and document the limitation on the config in the meantime. It sits naturally with the registry-lifetime work already tracked in #19759, which removes the process-wide map and makes per-writer scoping fall out of it. -- 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]
