hudi-agent commented on code in PR #19790:
URL: https://github.com/apache/hudi/pull/19790#discussion_r3900942015
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metrics/TestDistributedRegistry.java:
##########
@@ -228,4 +229,74 @@ public void testGetAllCountsWithPrefix() {
Assertions.assertTrue(countsWithoutPrefix.containsKey(METRIC_1));
Assertions.assertTrue(countsWithoutPrefix.containsKey(METRIC_2));
}
+
+ @Test
+ public void testRegisterIdempotent() {
+ // Given: a DistributedRegistry registered to the SparkContext
+ String registryName = REGISTRY_NAME + "_testIdempotent";
+ DistributedRegistry registry = new DistributedRegistry(registryName);
+ registry.add(METRIC_1, 42);
+ DistributedRegistry result = registry.register(jsc);
+
+ // Then: first registration returns the same instance
+ Assertions.assertSame(registry, result);
+ Assertions.assertTrue(registry.isRegistered());
+
+ // When: register() is called again on the same SparkContext
+ DistributedRegistry result2 = registry.register(jsc);
+
+ // Then: it is a no-op, returns same instance
+ Assertions.assertSame(registry, result2);
+ }
+
+ @Test
+ public void testRegisterHandlesStaleAccumulator() {
+ // Simulate the stale-singleton scenario: a DistributedRegistry that was
registered
+ // to a previous SparkContext and cannot be re-registered because
AccumulatorV2.metadata
+ // is already set (throws IllegalStateException).
+
+ // Given: a registry cached in the global REGISTRY_MAP
+ String registryName = REGISTRY_NAME + "_testStale";
+ DistributedRegistry staleRegistry = new DistributedRegistry(registryName);
+ staleRegistry.add(METRIC_1, 100);
+ staleRegistry.add(METRIC_2, 200);
+ String cacheKey = Registry.makeKey("", registryName);
+ Registry.REGISTRY_MAP.put(cacheKey, staleRegistry);
+
+ // Given: register to SparkContext #1
+ staleRegistry.register(jsc);
+ Assertions.assertTrue(staleRegistry.isRegistered());
+
+ // Simulate SparkContext restart: stop and create a new one
+ jsc.stop();
+ SparkConf conf = HoodieClientTestUtils.getSparkConfForTest(
+ TestDistributedRegistry.class.getSimpleName() + "_stale");
+ JavaSparkContext jsc2 = new JavaSparkContext(conf);
Review Comment:
🤖 nit: the comment here admits the catch path may never be hit in local
mode, which undercuts what the test name promises — could you either rename it
to something like `testRegisterWithNewSparkContext` that reflects what it
actually exercises, or add a note explaining how the stale-singleton catch
branch is separately validated?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/DistributedRegistry.java:
##########
@@ -45,9 +45,27 @@ public String getName() {
return name;
}
- public void register(JavaSparkContext jsc) {
- if (!isRegistered()) {
+ public DistributedRegistry register(JavaSparkContext jsc) {
Review Comment:
🤖 Since the whole point here is a JVM-wide singleton shared across multiple
write clients, could two threads race in this method? The `isRegistered()`
check and the catch/swap are unsynchronized, so two concurrent callers on the
same stale instance could each build a `fresh`, both call
`jsc.sc().register(fresh)` (registering two accumulators), and then race on the
`REGISTRY_MAP` swap — leaving one registered accumulator orphaned and callers
holding different instances. If concurrent write-client init is possible, would
it be worth synchronizing register() (or guarding the swap)? @nsivabalan
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metrics/DistributedRegistry.java:
##########
@@ -45,9 +45,27 @@ public String getName() {
return name;
}
- public void register(JavaSparkContext jsc) {
- if (!isRegistered()) {
+ public DistributedRegistry register(JavaSparkContext jsc) {
+ if (isRegistered()) {
+ return this;
+ }
+ try {
jsc.sc().register(this);
+ return this;
+ } catch (IllegalStateException e) {
+ // Stale singleton: was registered to a previous SparkContext that no
longer exists.
+ // AccumulatorV2.metadata is non-null (so register() rejects it) but
isRegistered()
+ // returned false (the id is not in the current AccumulatorContext).
+ // Create a fresh accumulator, register it, and swap it into the
registry cache.
+ DistributedRegistry fresh = new DistributedRegistry(this.name);
+ fresh.counters.putAll(this.counters);
+ jsc.sc().register(fresh);
+ Registry.REGISTRY_MAP.forEach((key, registry) -> {
Review Comment:
🤖 Agreed that `getMetricRegistry` is the natural spot — it's the only place
both caches are visible (`DISTRIBUTED_REGISTRY_MAP` directly, and
`Registry.REGISTRY_MAP` via `getRegistryOfClass`), and
`createWrapperFileSystemRegistries` funnels through it too, so a swap confined
to `DistributedRegistry.register()` can't reach `DISTRIBUTED_REGISTRY_MAP`. One
thing to watch though: the detect-and-replace has to run on the value returned
by `computeIfAbsent`, not inside the lambda — otherwise the short-circuit Danny
flagged means an already-cached stale entry is never re-evaluated. A staleness
check plus an explicit `put`/`compute` on the returned registry would cover
both caches and that computeIfAbsent path.
--
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]
