danny0405 commented on code in PR #19575:
URL: https://github.com/apache/hudi/pull/19575#discussion_r3877808409
##########
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:
I would go one step further and make the distributed-registry map owned by
the `HoodieSparkEngineContext` instance instead of adding another override mode
to the process-wide `Registry.REGISTRY_MAP`. That would hide the registry
details here while also removing the base-path digest, stale-SparkContext
replacement branch, test-only eviction helper, and the duplicate empty-table
alias installed by `setRegistries`.
It also creates a natural place to introduce per-writer or per-write
ownership. With the current table-scoped static accumulator, two write clients
in the same JVM can still clear and publish each other counters. This is the
same lifetime simplification tracked in #19759, but it looks structurally
cleaner than making `getRegistryOfClass` support replacement.
--
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]
