danny0405 commented on code in PR #19575:
URL: https://github.com/apache/hudi/pull/19575#discussion_r3877756560
##########
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:
Can we hide these registry details from spark context:
1. make `getRegistryOfClass` thread safe;
2. add a param or new API to support overide instead of existing
`computeIfAbsent`;
--
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]
