danny0405 commented on code in PR #19790:
URL: https://github.com/apache/hudi/pull/19790#discussion_r3891364442
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/HoodieSparkEngineContext.java:
##########
@@ -283,8 +283,7 @@ public Registry getMetricRegistry(String tableName, String
registryName) {
final String prefixedName = tableName.isEmpty() ? registryName : tableName
+ "." + registryName;
return DISTRIBUTED_REGISTRY_MAP.computeIfAbsent(prefixedName, key -> {
Registry registry = Registry.getRegistryOfClass(tableName, registryName,
DistributedRegistry.class.getName());
- ((DistributedRegistry) registry).register(javaSparkContext);
- return registry;
+ return ((DistributedRegistry) registry).register(javaSparkContext);
Review Comment:
[P1] Recover existing entries in the engine registry cache
`DISTRIBUTED_REGISTRY_MAP` is static and survives SparkContext restarts, so
putting the recovery call inside `computeIfAbsent` skips it for registries
already cached by the previous context. Under the stale-accumulator condition
this PR handles, a subsequent lookup therefore still returns the unregistered
instance. I reproduced this with the PR classes on Spark 3.5.5: obtain a
registry through the first engine context, restart Spark, remove the old
accumulator ID from `AccumulatorContext` to induce the stale state, and look up
the same registry through the new engine context. The lookup returns the old
instance with `isRegistered() == false`, and `engineContext.map(...)` fails
with `Task not serializable`, caused by `Accumulator must be registered before
send to executor`. Please scope/reset this cache with the SparkContext or
recover its stale entries before reuse; replacing only `Registry.REGISTRY_MAP`
does not update the map captured by engine operations.
##########
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);
+
+ try {
+ // When: register() is called with the new SparkContext
+ // In local mode, isRegistered() may still return true since
AccumulatorContext
+ // persists across stop/start. Force the stale path by calling register
on jsc2.
+ // The key invariant: it must not throw IllegalStateException.
+ DistributedRegistry result = staleRegistry.register(jsc2);
Review Comment:
[P2] Force the stale state so this test exercises recovery
Stopping and recreating the SparkContext does not remove this strongly
referenced accumulator from the JVM-wide `AccumulatorContext` on Spark 3.5.5.
Consequently, this call returns from the `isRegistered()` guard and never
exercises the catch block or cache replacement. I ran both added tests
successfully, then removed the entire recovery block while retaining the new
return type, and both tests still passed. Please explicitly remove
`staleRegistry.id()` from `AccumulatorContext` and assert
`!staleRegistry.isRegistered()` before calling `register()`. Then assert that
the result is a different instance and that
`Registry.REGISTRY_MAP.get(cacheKey)` is that result, alongside the counter
checks, so removing or breaking the fix actually fails the regression test.
--
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]
