yihua commented on code in PR #19575:
URL: https://github.com/apache/hudi/pull/19575#discussion_r3867084520
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkTable.java:
##########
@@ -142,9 +146,19 @@ protected Option<HoodieTableMetadataWriter>
getMetadataWriter(
return Option.empty();
}
+ /**
+ * Carries thread-local task state onto Hudi's writer pools. The metrics
binding travels with the
+ * {@link TaskContext} for the same reason that does: {@code
BoundedInMemoryExecutor} and
+ * {@code DisruptorExecutor} run on threads the task did not create. No
unbind, since these pools are
+ * created and shut down per operation.
+ */
@Override
public Runnable getPreExecuteRunnable() {
final TaskContext taskContext = TaskContext.get();
- return () -> TaskContext$.MODULE$.setTaskContext(taskContext);
+ final Map<String, Registry> metricsBinding =
ExecutorMetricsContext.capture();
Review Comment:
non-blocking: Tracing the three `bind` call sites, this one can only ever
capture an empty map — the two lookup functions bind and unbind within a single
`call()` (and `unbind(null)` does `BOUND.remove()`), and
`getPreExecuteRunnable` is only reached from the write stage
(`HoodieMergeHelper`, `SparkLazyInsertIterable`, the bootstrap handlers,
`SingleSparkJobConsistentHashingExecutionStrategy`), never from inside a
lookup. So the net effect is that `isBound()` becomes true on every writer pool
thread with nothing bound, which is exactly the case where
`Registry.getRegistry(name)` now returns `NoOpRegistry` instead of the
process-wide `LocalRegistry`. Would it be simpler to drop this hunk until
something below the write API actually emits, rather than carrying the flip for
a binding that is always empty?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/HoodieSparkEngineContext.java:
##########
@@ -278,11 +279,47 @@ 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 removeMetricRegistry(String tableName, String
registryName) {
+ String prefixedName = tableName.isEmpty() ? registryName : tableName + "."
+ registryName;
+ DISTRIBUTED_REGISTRY_MAP.remove(prefixedName);
+ Registry.REGISTRY_MAP.remove(Registry.makeKey(tableName, registryName));
+ // setRegistries also indexes under the empty table name, so evicting only
the table-scoped key leaves
+ // the accumulator reachable under ::<table>.<registry>.
+ Registry.REGISTRY_MAP.remove(Registry.makeKey("", prefixedName));
+ }
+
@Override
public Registry getMetricRegistry(String tableName, String registryName) {
final String prefixedName = tableName.isEmpty() ? registryName : tableName
+ "." + registryName;
- return DISTRIBUTED_REGISTRY_MAP.computeIfAbsent(prefixedName, key -> {
+ // Both maps are process-wide statics that outlive any SparkContext, so
the staleness check and the
+ // recreation have to be atomic: otherwise one caller can evict the
registry another caller just
+ // created, leaving two live accumulators for one metric name while
reporting only ever reads the
+ // one still in the map.
+ return DISTRIBUTED_REGISTRY_MAP.compute(prefixedName, (key, cached) -> {
+ if (cached instanceof DistributedRegistry && ((DistributedRegistry)
cached).isRegisteredWith(javaSparkContext)) {
+ return cached;
+ }
+ // Nothing usable cached, or the cached accumulator is bound to a
SparkContext that is no longer
+ // live (a restart in the same JVM: shells, notebooks, Spark Connect).
Drop the shared-map entry
+ // first, since getRegistryOfClass() would otherwise hand back that same
stale instance.
+ final String sharedKey = Registry.makeKey(tableName, registryName);
+ Registry.REGISTRY_MAP.remove(sharedKey);
Review Comment:
non-blocking: `removeMetricRegistry` above drops all three keys and its
comment explains why the `::<table>.<registry>` one matters, but this eviction
path only removes `makeKey(tableName, registryName)`. In local mode
`setRegistries` puts the registry into the driver's own `REGISTRY_MAP` under
`makeKey("", prefixedName)`, so after a SparkContext restart that entry still
points at the dead accumulator while the new one is registered under the
table-scoped key — and since both report the same prefixed names from
`getAllMetrics`, which value survives depends on map iteration order. Worth
adding the same `Registry.REGISTRY_MAP.remove(Registry.makeKey("",
prefixedName))` here?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/RecordIndexLookupMetrics.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.NoOpRegistry;
+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.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 {
+
+ private RecordIndexLookupMetrics() {
+ }
+
+ /**
+ * 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) {
+ Map<String, Registry> bundle = new HashMap<>();
+ for (ExecutorMetricRegistry metricRegistry :
ExecutorMetricRegistry.values()) {
+ if (!metricRegistry.isEnabled(config)) {
Review Comment:
Since the "follows `hoodie.metrics.on`" behaviour comes from the infer
function, an explicit `hoodie.metrics.rli.lookup.enable=true` with
`hoodie.metrics.on=false` still registers the accumulator and counts on every
task, but `HoodieMetrics.metrics` is null so `publishToReporter` returns at its
null guard and the counters are released unread each commit. Would it be worth
adding `config.isMetricsOn()` to the gate here, so the Impact section's
"nothing is collected and no accumulator is created" holds for the explicit
setting too?
--
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]
